diff --git a/docs/superpowers/specs/2026-07-23-unified-usb-camera-launch-design.md b/docs/superpowers/specs/2026-07-23-unified-usb-camera-launch-design.md index d9c3441..4a958d2 100644 --- a/docs/superpowers/specs/2026-07-23-unified-usb-camera-launch-design.md +++ b/docs/superpowers/specs/2026-07-23-unified-usb-camera-launch-design.md @@ -18,11 +18,12 @@ Add `config/usb_camera.yaml` as the single source of runtime camera parameters: - `io_method`: `mmap` - `zero_copy`: `false` - `frame_id`: `default_usb_cam` -- `camera_calibration_file_path`: the existing - `config/usb_camera_calibration.yaml` The calibration data remains in `usb_camera_calibration.yaml`; it is not mixed -with ROS runtime parameters. +with ROS runtime parameters. The shared base launch resolves its installed path +and applies it to the camera node after loading the runtime YAML. Both launch +variants therefore receive the same calibration path without a source-tree +absolute path. ## Launch Structure @@ -31,7 +32,9 @@ with ROS runtime parameters. 1. Declare one `camera_config` launch argument whose default is the installed `config/usb_camera.yaml` path. 2. Start the shared-memory environment once. -3. Start one `hobot_usb_cam` node with the YAML file as its parameter source. +3. Resolve the installed `usb_camera_calibration.yaml` path. +4. Start one `hobot_usb_cam` node with the YAML file as its parameter source and + the resolved calibration resource path. `hobot_usb_cam_websocket.launch.py` will: diff --git a/src/car_usb_cam/CMakeLists.txt b/src/car_usb_cam/CMakeLists.txt new file mode 100644 index 0000000..d3b1aea --- /dev/null +++ b/src/car_usb_cam/CMakeLists.txt @@ -0,0 +1,19 @@ +cmake_minimum_required(VERSION 3.8) +project(car_usb_cam) + +find_package(ament_cmake REQUIRED) + +install( + DIRECTORY config launch + DESTINATION share/${PROJECT_NAME} +) + +if(BUILD_TESTING) + find_package(ament_cmake_pytest REQUIRED) + ament_add_pytest_test( + test_unified_launch_config + test/test_unified_launch_config.py + ) +endif() + +ament_package() diff --git a/src/car_usb_cam/config/usb_camera.yaml b/src/car_usb_cam/config/usb_camera.yaml new file mode 100644 index 0000000..d6d6b49 --- /dev/null +++ b/src/car_usb_cam/config/usb_camera.yaml @@ -0,0 +1,10 @@ +hobot_usb_cam: + ros__parameters: + frame_id: default_usb_cam + framerate: 30 + image_height: 720 + image_width: 1280 + io_method: mmap + pixel_format: mjpeg + video_device: /dev/video0 + zero_copy: false diff --git a/src/car_usb_cam/launch/hobot_usb_cam.launch.py b/src/car_usb_cam/launch/hobot_usb_cam.launch.py index b01b7a7..2588b88 100644 --- a/src/car_usb_cam/launch/hobot_usb_cam.launch.py +++ b/src/car_usb_cam/launch/hobot_usb_cam.launch.py @@ -1,4 +1,4 @@ -# Copyright (c) 2024,D-Robotics. +# Copyright (c) 2024, D-Robotics. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -12,82 +12,46 @@ # See the License for the specific language governing permissions and # limitations under the License. -from launch import LaunchDescription -from launch.actions import DeclareLaunchArgument -from launch.substitutions import LaunchConfiguration, TextSubstitution -from launch_ros.actions import Node -from launch.actions import IncludeLaunchDescription -from launch.launch_description_sources import PythonLaunchDescriptionSource -from ament_index_python import get_package_share_directory -from ament_index_python.packages import get_package_prefix import os +from ament_index_python.packages import get_package_share_directory +from launch import LaunchDescription +from launch.actions import DeclareLaunchArgument, IncludeLaunchDescription +from launch.launch_description_sources import PythonLaunchDescriptionSource +from launch.substitutions import LaunchConfiguration +from launch_ros.actions import Node + + def generate_launch_description(): - config_file_path = os.path.join( - get_package_prefix('hobot_usb_cam'), - "/home/sunrise/yiliao_ws/src/car_usb_cam/config/usb_camera_calibration.yaml") - print("config_file_path is ", config_file_path) + package_share = get_package_share_directory("car_usb_cam") + camera_config = os.path.join(package_share, "config", "usb_camera.yaml") + calibration_file = os.path.join( + package_share, "config", "usb_camera_calibration.yaml" + ) return LaunchDescription([ DeclareLaunchArgument( - 'usb_camera_calibration_file_path', - default_value=TextSubstitution(text=str(config_file_path)), - description='camera calibration file path'), - DeclareLaunchArgument( - 'usb_frame_id', - default_value='default_usb_cam', - description='image message frame_id'), - DeclareLaunchArgument( - 'usb_framerate', - default_value='30', - description='framerate'), - DeclareLaunchArgument( - 'usb_image_height', - default_value='450', - description='image height'), - DeclareLaunchArgument( - 'usb_image_width', - default_value='800', - description='image width'), - DeclareLaunchArgument( - 'usb_io_method', - default_value='mmap', - description='io_method, mmap/read/userptr'), - DeclareLaunchArgument( - 'usb_pixel_format', - default_value='yuyv2rgb', - description='pixel format, mjpeg/yuyv2rgb'), - DeclareLaunchArgument( - 'usb_video_device', - default_value='/dev/video0', - description='usb camera device'), - DeclareLaunchArgument( - 'usb_zero_copy', - default_value='False', - description='use zero copy or not'), - # 启动零拷贝环境配置node + "camera_config", + default_value=camera_config, + description="ROS parameter file for the USB camera", + ), IncludeLaunchDescription( PythonLaunchDescriptionSource( os.path.join( - get_package_share_directory('hobot_shm'), - 'launch/hobot_shm.launch.py')) + get_package_share_directory("hobot_shm"), + "launch", + "hobot_shm.launch.py", + ) + ) ), Node( - package='hobot_usb_cam', - executable='hobot_usb_cam', - name='hobot_usb_cam', + package="hobot_usb_cam", + executable="hobot_usb_cam", + name="hobot_usb_cam", parameters=[ - {"camera_calibration_file_path": LaunchConfiguration( - 'usb_camera_calibration_file_path')}, - {"frame_id": LaunchConfiguration('usb_frame_id')}, - {"framerate": LaunchConfiguration('usb_framerate')}, - {"image_height": LaunchConfiguration('usb_image_height')}, - {"image_width": LaunchConfiguration('usb_image_width')}, - {"io_method": LaunchConfiguration('usb_io_method')}, - {"pixel_format": LaunchConfiguration('usb_pixel_format')}, - {"video_device": LaunchConfiguration('usb_video_device')}, - {"zero_copy": LaunchConfiguration('usb_zero_copy')} + LaunchConfiguration("camera_config"), + {"camera_calibration_file_path": calibration_file}, ], - arguments=['--ros-args', '--log-level', 'warn'] - ) + arguments=["--ros-args", "--log-level", "warn"], + ), ]) diff --git a/src/car_usb_cam/launch/hobot_usb_cam_websocket.launch.py b/src/car_usb_cam/launch/hobot_usb_cam_websocket.launch.py index eb9c3f9..8f76bd2 100644 --- a/src/car_usb_cam/launch/hobot_usb_cam_websocket.launch.py +++ b/src/car_usb_cam/launch/hobot_usb_cam_websocket.launch.py @@ -1,4 +1,4 @@ -# Copyright (c) 2024,D-Robotics. +# Copyright (c) 2024, D-Robotics. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -13,82 +13,38 @@ # limitations under the License. import os -from launch import LaunchDescription -from launch.actions import DeclareLaunchArgument -from launch.substitutions import LaunchConfiguration -from launch_ros.actions import Node +from ament_index_python.packages import get_package_share_directory +from launch import LaunchDescription from launch.actions import IncludeLaunchDescription from launch.launch_description_sources import PythonLaunchDescriptionSource -from ament_index_python import get_package_share_directory def generate_launch_description(): - camera_node = None - - print("using usb camera") - # using usb cam publish image - usb_cam_device_arg = DeclareLaunchArgument( - 'device', - default_value='/dev/video0', - description='usb camera device') - - usb_node = IncludeLaunchDescription( + camera_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( os.path.join( - get_package_share_directory('car_usb_cam'), - 'launch/hobot_usb_cam.launch.py')), - launch_arguments={ - 'usb_image_width': '800', - 'usb_image_height': '450', - 'usb_framerate': '30', - 'usb_pixel_format': 'yuyv2rgb',# yuyv2rgb - 'usb_zero_copy': 'False', - 'usb_video_device': LaunchConfiguration('device') - }.items() + get_package_share_directory("car_usb_cam"), + "launch", + "hobot_usb_cam.launch.py", + ) + ) ) - # nv12->jpeg - jpeg_codec_node = IncludeLaunchDescription( + websocket_launch = IncludeLaunchDescription( PythonLaunchDescriptionSource( os.path.join( - get_package_share_directory('hobot_codec'), - 'launch/hobot_codec_encode.launch.py')), - launch_arguments={ - #'codec_in_mode': 'shared_mem', - 'codec_in_mode': 'ros', - 'codec_in_format': 'rgb8',#rgb8 - 'codec_out_mode': 'ros', # - #'codec_sub_topic': '/hbmem_img', - 'codec_sub_topic': '/image', - 'codec_pub_topic': '/image_mjpeg' # 原image_mjpeg - }.items() - ) - # web - web_node = IncludeLaunchDescription( - PythonLaunchDescriptionSource( - os.path.join( - get_package_share_directory('websocket'), - 'launch/websocket.launch.py')), - launch_arguments={ - 'websocket_image_topic': '/image_mjpeg', - 'websocket_only_show_image': 'True' - }.items() - ) - - - return LaunchDescription([ - # 启动零拷贝环境配置node - IncludeLaunchDescription( - PythonLaunchDescriptionSource( - os.path.join( - get_package_share_directory('hobot_shm'), - 'launch/hobot_shm.launch.py')) + get_package_share_directory("websocket"), + "launch", + "websocket.launch.py", + ) ), - usb_cam_device_arg, - usb_node, - # image codec - jpeg_codec_node, - # web display - # web_node - ]) + launch_arguments={ + "websocket_image_topic": "/image", + "websocket_image_type": "mjpeg", + "websocket_only_show_image": "True", + "websocket_output_fps": "30", + }.items(), + ) + + return LaunchDescription([camera_launch, websocket_launch]) diff --git a/src/car_usb_cam/package.xml b/src/car_usb_cam/package.xml index 98e2b7d..976fc14 100644 --- a/src/car_usb_cam/package.xml +++ b/src/car_usb_cam/package.xml @@ -18,10 +18,18 @@ yaml_cpp_vendor ffmpeg + ament_index_python + launch + launch_ros + hobot_usb_cam + hobot_shm + websocket rosidl_interface_packages ament_lint_auto ament_lint_common + ament_cmake_pytest + python3-yaml ament_cmake diff --git a/src/car_usb_cam/test/test_unified_launch_config.py b/src/car_usb_cam/test/test_unified_launch_config.py new file mode 100644 index 0000000..998311a --- /dev/null +++ b/src/car_usb_cam/test/test_unified_launch_config.py @@ -0,0 +1,131 @@ +import ast +import pathlib +import unittest +import xml.etree.ElementTree as ET + +import yaml + + +PACKAGE_ROOT = pathlib.Path(__file__).resolve().parents[1] +BASE_LAUNCH = PACKAGE_ROOT / "launch" / "hobot_usb_cam.launch.py" +WEB_LAUNCH = PACKAGE_ROOT / "launch" / "hobot_usb_cam_websocket.launch.py" +CAMERA_CONFIG = PACKAGE_ROOT / "config" / "usb_camera.yaml" +PACKAGE_XML = PACKAGE_ROOT / "package.xml" + + +def string_constants(path): + tree = ast.parse(path.read_text(encoding="utf-8")) + return { + node.value + for node in ast.walk(tree) + if isinstance(node, ast.Constant) and isinstance(node.value, str) + } + + +def call_names(path): + tree = ast.parse(path.read_text(encoding="utf-8")) + names = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + if isinstance(node.func, ast.Name): + names.append(node.func.id) + elif isinstance(node.func, ast.Attribute): + names.append(node.func.attr) + return names + + +def calls_named(path, name): + tree = ast.parse(path.read_text(encoding="utf-8")) + calls = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + function_name = None + if isinstance(node.func, ast.Name): + function_name = node.func.id + elif isinstance(node.func, ast.Attribute): + function_name = node.func.attr + if function_name == name: + calls.append(node) + return calls + + +def assigned_call(path, variable_name): + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Call): + continue + if any( + isinstance(target, ast.Name) and target.id == variable_name + for target in node.targets + ): + return node.value + raise AssertionError(f"No call assigned to {variable_name}") + + +class UnifiedLaunchConfigTest(unittest.TestCase): + def test_yaml_defines_final_native_mjpeg_configuration(self): + document = yaml.safe_load(CAMERA_CONFIG.read_text(encoding="utf-8")) + params = document["hobot_usb_cam"]["ros__parameters"] + self.assertEqual(params["video_device"], "/dev/video0") + self.assertEqual(params["image_width"], 1280) + self.assertEqual(params["image_height"], 720) + self.assertEqual(params["framerate"], 30) + self.assertEqual(params["pixel_format"], "mjpeg") + self.assertEqual(params["io_method"], "mmap") + self.assertFalse(params["zero_copy"]) + self.assertNotIn("camera_calibration_file_path", params) + + def test_base_launch_owns_the_only_camera_node(self): + constants = string_constants(BASE_LAUNCH) + self.assertIn("usb_camera.yaml", constants) + self.assertIn("usb_camera_calibration.yaml", constants) + self.assertIn("hobot_usb_cam", constants) + node_calls = calls_named(BASE_LAUNCH, "Node") + self.assertEqual(len(node_calls), 1) + parameters = next( + keyword.value + for keyword in node_calls[0].keywords + if keyword.arg == "parameters" + ) + parameters_tree = ast.dump(parameters) + self.assertIn("camera_config", parameters_tree) + self.assertIn("camera_calibration_file_path", parameters_tree) + + def test_web_launch_only_adds_websocket(self): + constants = string_constants(WEB_LAUNCH) + self.assertIn("hobot_usb_cam.launch.py", constants) + self.assertIn("websocket.launch.py", constants) + self.assertEqual(call_names(WEB_LAUNCH).count("Node"), 0) + self.assertEqual(len(calls_named(WEB_LAUNCH, "IncludeLaunchDescription")), 2) + camera_include = assigned_call(WEB_LAUNCH, "camera_launch") + self.assertNotIn("launch_arguments", {item.arg for item in camera_include.keywords}) + for forbidden in ( + "hobot_codec", + "/image_mjpeg", + "yuyv2rgb", + "usb_image_width", + "usb_image_height", + "usb_pixel_format", + ): + self.assertNotIn(forbidden, constants) + + def test_manifest_declares_launch_and_test_dependencies(self): + root = ET.parse(PACKAGE_XML).getroot() + runtime_dependencies = {item.text for item in root.findall("exec_depend")} + self.assertTrue({ + "ament_index_python", + "launch", + "launch_ros", + "hobot_usb_cam", + "hobot_shm", + "websocket", + }.issubset(runtime_dependencies)) + test_dependencies = {item.text for item in root.findall("test_depend")} + self.assertIn("ament_cmake_pytest", test_dependencies) + self.assertIn("python3-yaml", test_dependencies) + + +if __name__ == "__main__": + unittest.main()