adaptions for Humble

launch/xacro_live_view.launch.py

@@ -16,12 +16,10 @@
 
 from launch import LaunchDescription
 from launch.actions import DeclareLaunchArgument
-from launch.substitutions import Command
-from launch.substitutions import LaunchConfiguration
-from launch.substitutions import PathJoinSubstitution
+from launch.substitutions import Command, LaunchConfiguration, PathJoinSubstitution
 from launch_ros.actions import Node
-from launch_ros.substitutions import ExecutableInPackage
-from launch_ros.substitutions import FindPackageShare
+from launch_ros.descriptions import ParameterValue
+from launch_ros.substitutions import ExecutableInPackage, FindPackageShare
 
 
 def generate_launch_description():
@@ -36,9 +34,9 @@ def generate_launch_description():
         DeclareLaunchArgument(
             name='rviz_config',
             description='Rviz config file.',
-            default_value=PathJoinSubstitution([
-                FindPackageShare('xacro_live'), 'rviz/view_robot.rviz'
-            ])
+            default_value=PathJoinSubstitution(
+                [FindPackageShare('xacro_live'), 'rviz/view_robot.rviz']
+            ),
         )
     )
 
@@ -49,9 +47,9 @@ def generate_launch_description():
             executable='xacro_live',
             name='xacro_live',
             output='screen',
-            parameters=[{
-                'xacro_file': LaunchConfiguration('xacro_file')
-            }]
+            parameters=[
+                {'xacro_file': LaunchConfiguration('xacro_file')},
+            ],
         )
     )
     launch_description.add_action(
@@ -60,13 +58,20 @@ def generate_launch_description():
             executable='robot_state_publisher',
             name='robot_state_publisher',
             output='screen',
-            parameters=[{
-                'robot_description':
-                Command([
-                    ExecutableInPackage(package='xacro', executable='xacro'), ' ',
-                    LaunchConfiguration('xacro_file')
-                ])
-            }]
+            parameters=[
+                {
+                    'robot_description': ParameterValue(
+                        Command(
+                            [
+                                ExecutableInPackage(package='xacro', executable='xacro'),
+                                ' ',
+                                LaunchConfiguration('xacro_file'),
+                            ]
+                        ),
+                        value_type=str,
+                    )
+                }
+            ],
         )
     )
     launch_description.add_action(
@@ -83,7 +88,7 @@ def generate_launch_description():
             executable='rviz2',
             name='rviz',
             output='screen',
-            arguments=['-d', LaunchConfiguration('rviz_config')]
+            arguments=['-d', LaunchConfiguration('rviz_config')],
         ),
     )
 

xacro_live/robot_description_client.py

@@ -20,12 +20,11 @@
 
 
 class RobotDescriptionClient:
-
     def __init__(
         self,
         client_node: rclpy.node.Node,
         server_node_name='robot_state_publisher',
-        param_name='robot_description'
+        param_name='robot_description',
     ):
         self.request = rcl_interfaces.srv.SetParameters.Request()
 
@@ -38,7 +37,7 @@ def __init__(
             rcl_interfaces.srv.SetParameters, server_node_name + '/set_parameters'
         )
 
-    def wait_for_service(self, timeout_sec=5.) -> None:
+    def wait_for_service(self, timeout_sec=5.0) -> None:
         if not self.client.wait_for_service(timeout_sec):
             raise RuntimeError('Wait for service timed out')
 

xacro_live/xacro_live_node.py

@@ -15,18 +15,21 @@
 import rclpy
 import rclpy.logging
 import rclpy.node
+from rcl_interfaces.msg import ParameterDescriptor
 
 from .xacro_observer import XacroObserver
-from .xacro_update_handler import RobotDescriptionClient
-from .xacro_update_handler import XacroUpdateHandler
+from .xacro_update_handler import RobotDescriptionClient, XacroUpdateHandler
 
 
-def main():
-
-    rclpy.init()
+def main(args=None):
+    rclpy.init(args=args)
     node = rclpy.create_node('xacro_live')
 
-    node.declare_parameter('xacro_file')
+    param_descriptor = ParameterDescriptor(
+        description='The file name of the URDF file relative to the ROS2 workspace.'
+    )
+    node.declare_parameter('xacro_file', '', param_descriptor)
+
     observer = XacroObserver(node.get_parameter('xacro_file').get_parameter_value().string_value)
     client = RobotDescriptionClient(node, 'robot_state_publisher')
     event_handler = XacroUpdateHandler(observer, client)

xacro_live/xacro_tree.py

@@ -18,7 +18,6 @@
 
 
 class XacroTree:
-
     def __init__(self, root_file: str):
         self.root_file = root_file
 
@@ -53,13 +52,14 @@ def update(self) -> None:
         """Process the xacro file and update files & directories."""
         xacro.all_includes = []
         self._doc = xacro.process_file(
-            self.root_file, **{
+            self.root_file,
+            **{
                 'output': None,
                 'just_deps': False,
                 'xacro_ns': True,
                 'verbosity': 1,
-                'mappings': {}
-            }
+                'mappings': {},
+            },
         )
         self._files = {os.path.realpath(file) for file in xacro.all_includes + [self.root_file]}
         self._dirs = {os.path.dirname(file) for file in self._files}

xacro_live/xacro_update_handler.py

@@ -13,16 +13,15 @@
 # limitations under the License.
 
 import rclpy.logging as roslog
+from launch_ros.descriptions import ParameterValue
 from rclpy.task import Future
-from watchdog.events import EVENT_TYPE_MODIFIED
-from watchdog.events import FileSystemEventHandler
+from watchdog.events import EVENT_TYPE_MODIFIED, FileSystemEventHandler
 
 from .robot_description_client import RobotDescriptionClient
 from .xacro_observer import XacroObserver
 
 
 class XacroUpdateHandler(FileSystemEventHandler):
-
     def __init__(self, xacro_observer: XacroObserver, client: RobotDescriptionClient):
         self.xacro_observer = xacro_observer
         self.logger = roslog.get_logger('xacro_live')
@@ -41,5 +40,5 @@ def on_modified(self, event):
                     )
                 except Exception as ex:  # noqa [flake8(B902)] TODO: specify exception types
                     self.future = Future()
-                    self.logger.warn('Invalid update!')
+                    self.logger.warn('on_modified: Invalid update!')
                     self.logger.warn(str(ex))