i/OrbbecSDK_ROS2
1
0
Fork 0
mirror of https://github.com/orbbec/OrbbecSDK_ROS2.git synced 2026-10-09 14:27:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add image transport plugin selection for images

Browse Source
This commit is contained in:
slz
2026-01-29 15:43:58 +08:00
committed by ob-yalian
parent 8ee48a63e0
commit bf0e5a0fd3
2 changed files with 115 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
orbbec_camera/launch/dabai_dcw2.launch.py
+84 -41
View File
@@ -1,5 +1,5 @@
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument
from launch.actions import DeclareLaunchArgument, OpaqueFunction
from launch.substitutions import LaunchConfiguration
from launch_ros.actions import PushRosNamespace
from launch.actions import GroupAction
@@ -7,7 +7,48 @@ from launch_ros.actions import ComposableNodeContainer
from launch_ros.descriptions import ComposableNode
from launch_ros.actions import Node
import os
import yaml
def convert_value(value):
if isinstance(value, str):
try:
return int(value)
except ValueError:
pass
try:
return float(value)
except ValueError:
pass
if value.lower() == 'true':
return True
elif value.lower() == 'false':
return False
return value
def load_parameters(context, args):
default_params = {arg.name: LaunchConfiguration(arg.name).perform(context) for arg in args}
skip_convert = {'config_file_path', 'usb_port', 'serial_number'}
result = {}
for key, value in default_params.items():
if key in skip_convert:
result[key] = value
elif 'enable_pub_plugins' in key:
if isinstance(value, str):
if value.startswith('[') and value.endswith(']'):
try:
result[key] = yaml.safe_load(value)
except:
result[key] = [value]
else:
result[key] = [value]
elif isinstance(value, list):
result[key] = value
else:
result[key] = [str(value)]
else:
result[key] = convert_value(value)
return result
def generate_launch_description():
# Declare arguments
@@ -82,54 +123,56 @@ def generate_launch_description():
DeclareLaunchArgument('laser_energy_level', default_value='-1'),
DeclareLaunchArgument('enable_heartbeat', default_value='false'),
DeclareLaunchArgument('industry_mode', default_value=''),
#color image transport plugins
DeclareLaunchArgument('color.image_raw.enable_pub_plugins',default_value='["image_transport/compressed", "image_transport/raw", "image_transport/theora"]'),
#depth image transport plugins
DeclareLaunchArgument('depth.image_raw.enable_pub_plugins',default_value='["image_transport/compressedDepth", "image_transport/raw"]'),
#ir image transport plugins
DeclareLaunchArgument('ir.image_raw.enable_pub_plugins',default_value='["image_transport/compressed", "image_transport/raw", "image_transport/theora"]'),
]
# Node configuration
parameters = [{arg.name: LaunchConfiguration(arg.name)} for arg in args]
# get ROS_DISTRO
ros_distro = os.environ["ROS_DISTRO"]
if ros_distro == "foxy":
return LaunchDescription(
args
+ [
def get_params(context, args):
return [load_parameters(context, args)]
def create_node_action(context, args):
params = get_params(context, args)
ros_distro = os.environ.get("ROS_DISTRO", "humble")
if ros_distro == "foxy":
return [
Node(
package="orbbec_camera",
executable="orbbec_camera_node",
name="ob_camera_node",
namespace=LaunchConfiguration("camera_name"),
parameters=parameters,
parameters=params,
output="screen",
)
]
)
# Define the ComposableNode
else:
# Define the ComposableNode
compose_node = ComposableNode(
package="orbbec_camera",
plugin="orbbec_camera::OBCameraNodeDriver",
name=LaunchConfiguration("camera_name"),
namespace="",
parameters=parameters,
)
# Define the ComposableNodeContainer
container = ComposableNodeContainer(
name="camera_container",
namespace="",
package="rclcpp_components",
executable="component_container",
composable_node_descriptions=[
compose_node,
],
output="screen",
)
# Launch description
ld = LaunchDescription(
args
+ [
GroupAction(
[PushRosNamespace(LaunchConfiguration("camera_name")), container]
)
else:
return [
GroupAction([
PushRosNamespace(LaunchConfiguration("camera_name")),
ComposableNodeContainer(
name="camera_container",
namespace="",
package="rclcpp_components",
executable="component_container",
composable_node_descriptions=[
ComposableNode(
package="orbbec_camera",
plugin="orbbec_camera::OBCameraNodeDriver",
name=LaunchConfiguration("camera_name"),
parameters=params,
),
],
output="screen",
)
])
]
)
return ld
return LaunchDescription(
args + [
OpaqueFunction(function=lambda context: create_node_action(context, args))
]
)
orbbec_camera/launch/gemini_330_series.launch.py
+31 -4
View File
@@ -43,10 +43,28 @@ def load_parameters(context, args):
yaml_params = load_yaml(config_file_path)
default_params = merge_params(default_params, yaml_params)
skip_convert = {'config_file_path', 'usb_port', 'serial_number'}
return {
key: (value if key in skip_convert else convert_value(value))
for key, value in default_params.items()
}
result = {}
for key, value in default_params.items():
if key in skip_convert:
result[key] = value
elif 'enable_pub_plugins' in key:
if isinstance(value, str):
if value.startswith('[') and value.endswith(']'):
try:
result[key] = yaml.safe_load(value)
except:
result[key] = [value]
else:
result[key] = [value]
elif isinstance(value, list):
result[key] = value
else:
result[key] = [str(value)]
else:
result[key] = convert_value(value)
return result
def generate_launch_description():
@@ -195,6 +213,15 @@ def generate_launch_description():
DeclareLaunchArgument('enable_hardware_reset', default_value='false'),
DeclareLaunchArgument('frame_aggregate_mode', default_value='ANY'), # full_frame、color_frame、ANY or disable
#color image transport plugins
DeclareLaunchArgument('color.image_raw.enable_pub_plugins',default_value='["image_transport/compressed", "image_transport/raw", "image_transport/theora"]'),
#depth image transport plugins
DeclareLaunchArgument('depth.image_raw.enable_pub_plugins',default_value='["image_transport/compressedDepth", "image_transport/raw"]'),
#infra1
DeclareLaunchArgument('left_ir.image_raw.enable_pub_plugins',default_value='["image_transport/compressed", "image_transport/raw", "image_transport/theora"]'),
#infra2
DeclareLaunchArgument('right_ir.image_raw.enable_pub_plugins',default_value='["image_transport/compressed", "image_transport/raw", "image_transport/theora"]'),
]
def get_params(context, args):