zbw/yiliao2026
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: unify usb camera launch configuration

Browse Source
This commit is contained in:
Orange
2026-07-23 15:36:21 +08:00
parent 51a3b3ff0b
commit 64d8ac6c44
7 changed files with 228 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/car_usb_cam/CMakeLists.txt Normal file
View File

@@ -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()

10
src/car_usb_cam/config/usb_camera.yaml Normal file
View File

@@ -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

96
src/car_usb_cam/launch/hobot_usb_cam.launch.py
View File

@@ -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"],
),
])

90
src/car_usb_cam/launch/hobot_usb_cam_websocket.launch.py
View File

@@ -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])

8
src/car_usb_cam/package.xml
View File

@@ -18,10 +18,18 @@
<depend>yaml_cpp_vendor</depend>
<!-- Only required for MJPEG to RGB converison -->
<depend>ffmpeg</depend>
<exec_depend>ament_index_python</exec_depend>
<exec_depend>launch</exec_depend>
<exec_depend>launch_ros</exec_depend>
<exec_depend>hobot_usb_cam</exec_depend>
<exec_depend>hobot_shm</exec_depend>
<exec_depend>websocket</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>
<test_depend>ament_lint_auto</test_depend>
<test_depend>ament_lint_common</test_depend>
<test_depend>ament_cmake_pytest</test_depend>
<test_depend>python3-yaml</test_depend>
<export>
<build_type>ament_cmake</build_type>

131
src/car_usb_cam/test/test_unified_launch_config.py Normal file
View File

@@ -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()