ruff

launch/ros2_video_streamer_node_launch.py

@@ -1,9 +1,18 @@
 from typing import Any, Dict, Tuple
 
-from launch.actions import DeclareLaunchArgument, OpaqueFunction
-from launch.launch_context import LaunchContext
-from launch.launch_description import LaunchDescription
-from launch.substitutions import LaunchConfiguration
+from launch.actions import (
+    DeclareLaunchArgument,
+    OpaqueFunction,
+)
+from launch.launch_context import (
+    LaunchContext,
+)
+from launch.launch_description import (
+    LaunchDescription,
+)
+from launch.substitutions import (
+    LaunchConfiguration,
+)
 from launch_ros.actions import Node
 
 
@@ -16,7 +25,9 @@ def generate_launch_description() -> LaunchDescription:
 
 # TODO remove args and kwargs?
 def launch_setup(
-    context: LaunchContext, *args: Tuple[Any], **kwargs: Dict[Any, Any]
+    context: LaunchContext,
+    *args: Tuple[Any],
+    **kwargs: Dict[Any, Any],
 ) -> LaunchDescription:
     """Generate array to be included in launch description."""
     # Declare and early evaluate camera_name argument
@@ -32,12 +43,17 @@ def launch_setup(
     camera_name_str = LaunchConfiguration("camera_name").perform(context)
 
     launch_arguments: list[DeclareLaunchArgument] = [
-        DeclareLaunchArgument("node_name", default_value="streamer_node"),
         DeclareLaunchArgument(
-            "image_topic_name", default_value=f"/{camera_name_str}/image_raw"
+            "node_name",
+            default_value="streamer_node",
         ),
         DeclareLaunchArgument(
-            "info_topic_name", default_value=f"/{camera_name_str}/camera_info"
+            "image_topic_name",
+            default_value=f"/{camera_name_str}/image_raw",
+        ),
+        DeclareLaunchArgument(
+            "info_topic_name",
+            default_value=f"/{camera_name_str}/camera_info",
         ),
         DeclareLaunchArgument(
             "config_file_name",
@@ -52,7 +68,8 @@ def launch_setup(
             description="`frame_id` field in the `CameraInfo` topic",
         ),
         DeclareLaunchArgument(
-            "type", description="Type of media source, (e.g. image or video)"
+            "type",
+            description="Type of media source, (e.g. image or video)",
         ),
         DeclareLaunchArgument(
             "file_name",
@@ -83,4 +100,10 @@ def launch_setup(
         ],
     )
 
-    return LaunchDescription([camera_name_argument, *launch_arguments, streamer_node])
+    return LaunchDescription(
+        [
+            camera_name_argument,
+            *launch_arguments,
+            streamer_node,
+        ]
+    )

launch/three_camera_launch.py

@@ -1,9 +1,15 @@
 import os
 
 import launch
-from ament_index_python.packages import get_package_share_directory
-from launch.actions import IncludeLaunchDescription
-from launch.launch_description_sources import PythonLaunchDescriptionSource
+from ament_index_python.packages import (
+    get_package_share_directory,
+)
+from launch.actions import (
+    IncludeLaunchDescription,
+)
+from launch.launch_description_sources import (
+    PythonLaunchDescriptionSource,
+)
 
 
 def generate_launch_description() -> launch.launch_description.LaunchDescription:
@@ -17,11 +23,17 @@ def generate_launch_description() -> launch.launch_description.LaunchDescription
     bottom_cam_launcher: IncludeLaunchDescription = create_cam_launcher("bottom")
 
     return launch.launch_description.LaunchDescription(
-        [manip_cam_launcher, front_cam_launcher, bottom_cam_launcher]
+        [
+            manip_cam_launcher,
+            front_cam_launcher,
+            bottom_cam_launcher,
+        ]
     )
 
 
-def create_cam_launcher(camera_name: str) -> IncludeLaunchDescription:
+def create_cam_launcher(
+    camera_name: str,
+) -> IncludeLaunchDescription:
     return IncludeLaunchDescription(
         PythonLaunchDescriptionSource(
             [
@@ -36,9 +48,21 @@ def create_cam_launcher(camera_name: str) -> IncludeLaunchDescription:
         # confused and makes them the same when we launch simultaneously
         launch_arguments=[
             ("type", "video"),
-            ("camera_name", f"{camera_name}_cam"),
-            ("image_topic_name", f"/{camera_name}_cam/image_raw"),
-            ("info_topic_name", f"/{camera_name}_cam/camera_info"),
-            ("file_name", f"{camera_name}_cam.mp4"),
+            (
+                "camera_name",
+                f"{camera_name}_cam",
+            ),
+            (
+                "image_topic_name",
+                f"/{camera_name}_cam/image_raw",
+            ),
+            (
+                "info_topic_name",
+                f"/{camera_name}_cam/camera_info",
+            ),
+            (
+                "file_name",
+                f"{camera_name}_cam.mp4",
+            ),
         ],
     )

ros2_video_streamer/ros2_video_streamer_node.py

@@ -2,21 +2,31 @@
 
 import cv2
 import rclpy
-from ament_index_python.packages import get_package_share_directory
+from ament_index_python.packages import (
+    get_package_share_directory,
+)
 from builtin_interfaces.msg import Time
 from cv2 import VideoCapture
 from cv2.typing import MatLike
 from cv_bridge import CvBridge
 from rclpy.node import Node
-from rclpy.qos import qos_profile_system_default
-from sensor_msgs.msg import CameraInfo, Image
+from rclpy.qos import (
+    qos_profile_system_default,
+)
+from sensor_msgs.msg import (
+    CameraInfo,
+    Image,
+)
 
 
 class VideoStreamerNode(Node):
     """ROS Camera simulator Node; reads video file & pubs ROS Images."""
 
     def __init__(self) -> None:
-        super().__init__("ros2_video_streamer_temp_name", parameter_overrides=[])
+        super().__init__(
+            "ros2_video_streamer_temp_name",
+            parameter_overrides=[],
+        )
 
         self.load_launch_parameters()
 
@@ -30,33 +40,54 @@ def __init__(self) -> None:
 
         # Publishers
         self.image_publisher_ = self.create_publisher(
-            Image, self.image_topic_name, qos_profile_system_default
+            Image,
+            self.image_topic_name,
+            qos_profile_system_default,
         )
         self.camera_info_publisher_ = self.create_publisher(
-            CameraInfo, self.info_topic_name, qos_profile_system_default
+            CameraInfo,
+            self.info_topic_name,
+            qos_profile_system_default,
         )
 
         if not os.path.isfile(self.path):
             raise RuntimeError(f"Invalid video path: {self.path}")
 
         if self.type == "video":
             self.vc: VideoCapture = cv2.VideoCapture(self.path)
-            self.vc.set(cv2.CAP_PROP_POS_MSEC, self.start)
+            self.vc.set(
+                cv2.CAP_PROP_POS_MSEC,
+                self.start,
+            )
             video_fps: float = self.vc.get(cv2.CAP_PROP_FPS)
         elif self.type == "image":
             self.image = cv2.imread(self.path)
             video_fps = 10
         else:
             raise ValueError(f"Unknown type: {self.type}")
 
-        self.timer = self.create_timer(1.0 / video_fps, self.image_callback)
+        self.timer = self.create_timer(
+            1.0 / video_fps,
+            self.image_callback,
+        )
         self.get_logger().info(f"Publishing image at {video_fps} fps")
 
-    def load_launch_parameters(self) -> None:
+    def load_launch_parameters(
+        self,
+    ) -> None:
         """Load the launch ROS parameters."""
-        self.declare_parameter("image_topic_name", value="/simulated_cam/image_raw")
-        self.declare_parameter("info_topic_name", value="/simulated_cam/camera_info")
-        self.declare_parameter("file_name", value="simulated_cam.mp4")
+        self.declare_parameter(
+            "image_topic_name",
+            value="/simulated_cam/image_raw",
+        )
+        self.declare_parameter(
+            "info_topic_name",
+            value="/simulated_cam/camera_info",
+        )
+        self.declare_parameter(
+            "file_name",
+            value="simulated_cam.mp4",
+        )
         # self.declare_parameter('config_file_path', value='')
         self.declare_parameter("loop", value=True)
         self.declare_parameter("frame_id", value="")
@@ -69,20 +100,17 @@ def load_launch_parameters(self) -> None:
         self.info_topic_name = (
             self.get_parameter("info_topic_name").get_parameter_value().string_value
         )
-        self.file_name = (
-            self.get_parameter("file_name").get_parameter_value().string_value
-        )
+        self.file_name = self.get_parameter("file_name").get_parameter_value().string_value
         # self.config_file_path = self.get_parameter('config_file_path')\
         #     .get_parameter_value().string_value
         self.loop = self.get_parameter("loop").get_parameter_value().bool_value
-        self.frame_id_ = (
-            self.get_parameter("frame_id").get_parameter_value().string_value
-        )
+        self.frame_id_ = self.get_parameter("frame_id").get_parameter_value().string_value
         self.type = self.get_parameter("type").get_parameter_value().string_value
         self.start = self.get_parameter("start").get_parameter_value().integer_value
 
         self.path = os.path.join(
-            get_package_share_directory("ros2_video_streamer"), self.file_name
+            get_package_share_directory("ros2_video_streamer"),
+            self.file_name,
         )
 
     # def load_config_file(self, file_path: str):
@@ -123,8 +151,14 @@ def image_callback(self) -> None:
                 self.destroy_node()
                 exit()
             elif not rval and self.loop:
-                self.vc.set(cv2.CAP_PROP_POS_MSEC, 0)
-                rval, image = self.vc.read()
+                self.vc.set(
+                    cv2.CAP_PROP_POS_MSEC,
+                    0,
+                )
+                (
+                    rval,
+                    image,
+                ) = self.vc.read()
         elif self.type == "image":
             image = self.image
         else:

setup.py

@@ -3,7 +3,10 @@
 import os
 from glob import glob
 
-from setuptools import find_packages, setup
+from setuptools import (
+    find_packages,
+    setup,
+)
 
 PACKAGE_NAME = "ros2_video_streamer"
 
@@ -12,15 +15,42 @@
     version="1.1.0",
     packages=find_packages(),
     data_files=[
-        ("share/ament_index/resource_index/packages", ["resource/" + PACKAGE_NAME]),
-        ("share/" + PACKAGE_NAME, ["package.xml"]),
-        (os.path.join("share", PACKAGE_NAME, "launch"), glob("launch/*.launch.xml")),
         (
-            os.path.join("share", PACKAGE_NAME, "launch"),
+            "share/ament_index/resource_index/packages",
+            ["resource/" + PACKAGE_NAME],
+        ),
+        (
+            "share/" + PACKAGE_NAME,
+            ["package.xml"],
+        ),
+        (
+            os.path.join(
+                "share",
+                PACKAGE_NAME,
+                "launch",
+            ),
+            glob("launch/*.launch.xml"),
+        ),
+        (
+            os.path.join(
+                "share",
+                PACKAGE_NAME,
+                "launch",
+            ),
             glob("launch/*launch.[pxy][yma]*"),
         ),
-        (os.path.join("share", PACKAGE_NAME, "config"), glob("config/*.yaml")),
-        (os.path.join("share", PACKAGE_NAME), glob("*.mp4")),
+        (
+            os.path.join(
+                "share",
+                PACKAGE_NAME,
+                "config",
+            ),
+            glob("config/*.yaml"),
+        ),
+        (
+            os.path.join("share", PACKAGE_NAME),
+            glob("*.mp4"),
+        ),
     ],
     py_modules=[],
     zip_safe=True,

test/test_flake8.py

@@ -15,14 +15,19 @@
 # limitations under the License.
 
 import pytest
-from ament_flake8.main import main_with_errors
+from ament_flake8.main import (
+    main_with_errors,
+)
 
 
 @pytest.mark.flake8
 @pytest.mark.linter
 def test_flake8() -> None:
     """Tests flake8 on this module."""
-    error_code, errors = main_with_errors(argv=[])
-    assert (
-        error_code == 0
-    ), f"Found {len(errors)} code style errors / warnings:\n" + "\n".join(errors)
+    (
+        error_code,
+        errors,
+    ) = main_with_errors(argv=[])
+    assert error_code == 0, f"Found {len(errors)} code style errors / warnings:\n" + "\n".join(
+        errors
+    )

test/test_mypy.py

@@ -11,7 +11,14 @@
 def test_mypy() -> None:
     """Tests mypy on this module."""
     file_path = __file__.replace(f"{__name__}.py", "")
-    config_file = os.path.join(file_path, "..", "..", "..", "..", ".mypy.ini")
+    config_file = os.path.join(
+        file_path,
+        "..",
+        "..",
+        "..",
+        "..",
+        ".mypy.ini",
+    )
     if not os.path.exists(config_file):
         config_file = os.path.join(file_path, "..", ".mypy.ini")
     error_code = main(argv=["--config", config_file])