Cleanup lifecycle_py to conform to ROS 2 standards.

lifecycle_py/lifecycle_py/talker.py

@@ -30,127 +30,128 @@
 
 
 class LifecycleTalker(Node):
-  """Our lifecycle talker node."""
-
-  def __init__(self, node_name, **kwargs):
-    """Construct the node."""
-    self._count: int = 0
-    self._pub: Optional[Publisher] = None
-    self._timer: Optional[Timer] = None
-    super().__init__(node_name, **kwargs)
-
-  def publish(self):
-    """Publish a new message when enabled."""
-    msg = std_msgs.msg.String()
-    msg.data = "Lifecycle HelloWorld #" + str(self._count);
-    self._count += 1
-
-    # Print the current state for demo purposes
-    if self._pub is None or not self._pub.is_activated:
-      self.get_logger().info('Lifecycle publisher is currently inactive. Messages are not published.');
-    else:
-      self.get_logger().info(f'Lifecycle publisher is active. Publishing: [{msg.data}]');
-
-    # We independently from the current state call publish on the lifecycle
-    # publisher.
-    # Only if the publisher is in an active state, the message transfer is
-    # enabled and the message actually published.
-    if self._pub is not None:
-      self._pub.publish(msg)
-
-  def on_configure(self, state: State) -> TransitionCallbackReturn:
-    """
-    Configure the node, after a configuring transition is requested.
-
-    on_configure callback is being called when the lifecycle node
-    enters the "configuring" state.
-    
-    :return: The state machine either invokes a transition to the "inactive" state or stays
-    in "unconfigured" depending on the return value.
-      TransitionCallbackReturn.SUCCESS transitions to "inactive".
-      TransitionCallbackReturn.FAILURE transitions to "unconfigured".
-      TransitionCallbackReturn.ERROR or any uncaught exceptions to "errorprocessing"
-    """
-    self._pub = self.create_lifecycle_publisher(std_msgs.msg.String, "lifecycle_chatter", 10);
-    self._timer_ = self.create_timer(1.0, self.publish)
-
-    self.get_logger().info("on_configure() is called.")
-    return TransitionCallbackReturn.SUCCESS
-
-  def on_activate(self, state: State) -> TransitionCallbackReturn:
-    # Differently to rclcpp, a lifecycle publisher transitions automatically between the inactive and
-    # enabled state and viceversa.
-    # For that reason, we only need to write an on_configure() and on_cleanup() callbacks, and we don't
-    # need to write on_activate()/on_deactivate() callbacks.
-
-    # Log, only for demo purposes
-    self.get_logger().info("on_activate() is called.")
-
-    # The default LifecycleNode callback is the one transitioning
-    # LifecyclePublisher entities from inactive to enabled.
-    # If you override on_activate(), don't forget to call the parent class method as well!!
-    return super().on_activate(state)
-  
-  def on_deactivate(self, state: State) -> TransitionCallbackReturn:
-    # Log, only for demo purposes
-    self.get_logger().info("on_deactivate() is called.")
-    # Same reasong here that for on_activate().
-    # These are the two only cases where you need to call the parent method.
-    return super().on_deactivate(state)
-
-  def on_cleanup(self, state: State) -> TransitionCallbackReturn:
-    """
-    Cleanup the node, after a cleaning-up transition is requested.
-
-    on_cleanup callback is being called when the lifecycle node
-    enters the "cleaning up" state.
-    
-    :return: The state machine either invokes a transition to the "unconfigured" state or stays
-    in "inactive" depending on the return value.
-      TransitionCallbackReturn.SUCCESS transitions to "unconfigured".
-      TransitionCallbackReturn.FAILURE transitions to "inactive".
-      TransitionCallbackReturn.ERROR or any uncaught exceptions to "errorprocessing"
-    """
-    self.destroy_timer(self._timer)
-    self.destroy_publisher(self._pub)
-
-    self.get_logger().info('on_cleanup() is called.')
-    return TransitionCallbackReturn.SUCCESS
-
-  def on_shutdown(self, state: State) -> TransitionCallbackReturn:
-    """
-    Shutdown the node, after a shutting-down transition is requested.
-
-    on_shutdown callback is being called when the lifecycle node
-    enters the "shutting down" state.
-    
-    :return: The state machine either invokes a transition to the "finalized" state or stays
-    in the current state depending on the return value.
-      TransitionCallbackReturn.SUCCESS transitions to "unconfigured".
-      TransitionCallbackReturn.FAILURE transitions to "inactive".
-      TransitionCallbackReturn.ERROR or any uncaught exceptions to "errorprocessing"
-    """
-    self.destroy_timer(self._timer)
-    self.destroy_publisher(self._pub)
-
-    self.get_logger().info('on_shutdown() is called.')
-    return TransitionCallbackReturn.SUCCESS
+    """Our lifecycle talker node."""
+
+    def __init__(self, node_name, **kwargs):
+        """Construct the node."""
+        self._count: int = 0
+        self._pub: Optional[Publisher] = None
+        self._timer: Optional[Timer] = None
+        super().__init__(node_name, **kwargs)
+
+    def publish(self):
+        """Publish a new message when enabled."""
+        msg = std_msgs.msg.String()
+        msg.data = 'Lifecycle HelloWorld #' + str(self._count)
+        self._count += 1
+
+        # Print the current state for demo purposes
+        if self._pub is None or not self._pub.is_activated:
+            self.get_logger().info('Lifecycle publisher is inactive. Messages are not published.')
+        else:
+            self.get_logger().info(f'Lifecycle publisher is active. Publishing: [{msg.data}]')
+
+        # We independently from the current state call publish on the lifecycle
+        # publisher.
+        # Only if the publisher is in an active state, the message transfer is
+        # enabled and the message actually published.
+        if self._pub is not None:
+            self._pub.publish(msg)
+
+    def on_configure(self, state: State) -> TransitionCallbackReturn:
+        """
+        Configure the node, after a configuring transition is requested.
+
+        on_configure callback is being called when the lifecycle node
+        enters the "configuring" state.
+
+        :return: The state machine either invokes a transition to the "inactive" state or stays
+        in "unconfigured" depending on the return value.
+          TransitionCallbackReturn.SUCCESS transitions to "inactive".
+          TransitionCallbackReturn.FAILURE transitions to "unconfigured".
+          TransitionCallbackReturn.ERROR or any uncaught exceptions to "errorprocessing"
+        """
+        self._pub = self.create_lifecycle_publisher(std_msgs.msg.String, 'lifecycle_chatter', 10)
+        self._timer = self.create_timer(1.0, self.publish)
+
+        self.get_logger().info('on_configure() is called.')
+        return TransitionCallbackReturn.SUCCESS
+
+    def on_activate(self, state: State) -> TransitionCallbackReturn:
+        # Differently to rclcpp, a lifecycle publisher transitions automatically between the
+        # inactive and enabled state and viceversa.
+        # For that reason, we only need to write an on_configure() and on_cleanup() callbacks,
+        # and we don't need to write on_activate()/on_deactivate() callbacks.
+
+        # Log, only for demo purposes
+        self.get_logger().info('on_activate() is called.')
+
+        # The default LifecycleNode callback is the one transitioning
+        # LifecyclePublisher entities from inactive to enabled.
+        # If you override on_activate(), don't forget to call the parent class method as well!!
+        return super().on_activate(state)
+
+    def on_deactivate(self, state: State) -> TransitionCallbackReturn:
+        # Log, only for demo purposes
+        self.get_logger().info('on_deactivate() is called.')
+        # Same reasong here that for on_activate().
+        # These are the two only cases where you need to call the parent method.
+        return super().on_deactivate(state)
+
+    def on_cleanup(self, state: State) -> TransitionCallbackReturn:
+        """
+        Cleanup the node, after a cleaning-up transition is requested.
+
+        on_cleanup callback is being called when the lifecycle node
+        enters the "cleaning up" state.
+
+        :return: The state machine either invokes a transition to the "unconfigured" state or stays
+        in "inactive" depending on the return value.
+          TransitionCallbackReturn.SUCCESS transitions to "unconfigured".
+          TransitionCallbackReturn.FAILURE transitions to "inactive".
+          TransitionCallbackReturn.ERROR or any uncaught exceptions to "errorprocessing"
+        """
+        self.destroy_timer(self._timer)
+        self.destroy_publisher(self._pub)
+
+        self.get_logger().info('on_cleanup() is called.')
+        return TransitionCallbackReturn.SUCCESS
+
+    def on_shutdown(self, state: State) -> TransitionCallbackReturn:
+        """
+        Shutdown the node, after a shutting-down transition is requested.
+
+        on_shutdown callback is being called when the lifecycle node
+        enters the "shutting down" state.
+
+        :return: The state machine either invokes a transition to the "finalized" state or stays
+        in the current state depending on the return value.
+          TransitionCallbackReturn.SUCCESS transitions to "unconfigured".
+          TransitionCallbackReturn.FAILURE transitions to "inactive".
+          TransitionCallbackReturn.ERROR or any uncaught exceptions to "errorprocessing"
+        """
+        self.destroy_timer(self._timer)
+        self.destroy_publisher(self._pub)
+
+        self.get_logger().info('on_shutdown() is called.')
+        return TransitionCallbackReturn.SUCCESS
 
 
 # A lifecycle node has the same node API
 # as a regular node. This means we can spawn a
 # node, give it a name and add it to the executor.
 
 def main():
-  rclpy.init()
+    rclpy.init()
+
+    executor = rclpy.executors.SingleThreadedExecutor()
+    lc_node = LifecycleTalker('lc_talker')
+    executor.add_node(lc_node)
+    try:
+        executor.spin()
+    except (KeyboardInterrupt, rclpy.executors.ExternalShutdownException):
+        lc_node.destroy_node()
 
-  executor = rclpy.executors.SingleThreadedExecutor()
-  lc_node = LifecycleTalker('lc_talker')
-  executor.add_node(lc_node)
-  try:
-    executor.spin()
-  except (KeyboardInterrupt, rclpy.executors.ExternalShutdownException):
-    lc_node.destroy_node()
 
 if __name__ == '__main__':
-  main()
+    main()

lifecycle_py/test/test_copyright.py

@@ -0,0 +1,23 @@
+# Copyright 2015 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ament_copyright.main import main
+import pytest
+
+
+@pytest.mark.copyright
+@pytest.mark.linter
+def test_copyright():
+    rc = main(argv=['.', 'test'])
+    assert rc == 0, 'Found errors'

lifecycle_py/test/test_flake8.py

@@ -0,0 +1,25 @@
+# Copyright 2017 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ament_flake8.main import main_with_errors
+import pytest
+
+
+@pytest.mark.flake8
+@pytest.mark.linter
+def test_flake8():
+    rc, errors = main_with_errors(argv=[])
+    assert rc == 0, \
+        'Found %d code style errors / warnings:\n' % len(errors) + \
+        '\n'.join(errors)

lifecycle_py/test/test_pep257.py

@@ -0,0 +1,23 @@
+# Copyright 2015 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ament_pep257.main import main
+import pytest
+
+
+@pytest.mark.linter
+@pytest.mark.pep257
+def test_pep257():
+    rc = main(argv=['.', 'test'])
+    assert rc == 0, 'Found code style errors / warnings'