Zultron/2022 09 19 rebase for PRs: 2 minor ros

hw_device_mgr/logging/__init__.py

@@ -0,0 +1,45 @@
+import logging
+
+
+class Logging:
+    """Wrapper for `logging` module."""
+
+    _logging_class = logging
+
+    def __init__(self, name):
+        lc = self._logging_class
+        lo = self._logger = lc.getLogger(name)
+        lh = self._log_handler = lc.StreamHandler()
+        lf = lc.Formatter("%(asctime)s [%(levelname)s]%(name)s: %(message)s")
+        lh.setFormatter(lf)
+        lo.addHandler(lh)
+
+    # Translate Python str levels to str levels (int levels are the same)
+    _level_map = dict(
+        fatal="fatal",
+        error="error",
+        warning="warning",
+        info="info",
+        debug="debug",
+        notset="notset",
+    )
+
+    def setLevel(self, level):
+        if isinstance(level, str):
+            level = self._level_map.get(level.upper(), level).upper()
+        self._logger.setLevel(level)
+
+    def getLevel(self):
+        return self._logger.getEffectiveLevel()
+
+    def __getattr__(self, name):
+        if name in self._level_map:
+            return getattr(self._logger, self._level_map[name])
+        if name.lower() in self._level_map:
+            attr = self._level_map[name.lower()].upper()
+            return getattr(self._logging_class, attr)
+        raise AttributeError(f"'Logging' object has no attribute '{name}'")
+
+    @classmethod
+    def getLogger(cls, name):
+        return cls(name)

hw_device_mgr/logging/ros.py

@@ -0,0 +1,30 @@
+from rclpy import logging
+from . import Logging
+from rclpy.logging import LoggingSeverity
+
+
+class ROSLogging(Logging):
+    """Wrapper for `rclpy.logging` object."""
+
+    def __init__(self, name):
+        self._logger = logging.get_logger(name)
+
+    # Translate Python str levels to rclpy str levels (int levels are
+    # the same)
+    _rclpy_level_map = dict(
+        critical=LoggingSeverity.FATAL,
+        error=LoggingSeverity.ERROR,
+        warning=LoggingSeverity.WARN,
+        info=LoggingSeverity.INFO,
+        debug=LoggingSeverity.DEBUG,
+        notset=LoggingSeverity.UNSET,
+    )
+
+    def setLevel(self, level):
+        if isinstance(level, str):
+            level = self._rclpy_level_map.get(level.lower(), None)
+            assert level is not None, f"Invalid log level '{level}'"
+        self._logger.set_level(level)
+
+    def getLevel(self):
+        return self._logger.get_effective_level()

hw_device_mgr/logging/tests/test_logging.py

@@ -0,0 +1,21 @@
+import pytest
+from .. import Logging
+
+
+class TestLogging:
+    tc = Logging
+
+    @pytest.fixture
+    def obj(self):
+        return self.tc("test")
+
+    def test_init(self, obj):
+        assert hasattr(obj, "_logger")
+
+    def test_getLevel_setLevel(self, obj):
+        obj.setLevel("error")
+        assert obj.getLevel() == obj.ERROR
+
+    def test_getLogger(self):
+        logger = self.tc.getLogger("test")
+        assert isinstance(logger, self.tc)

hw_device_mgr/logging/tests/test_ros_logging.py

@@ -0,0 +1,13 @@
+import pytest
+
+try:
+    import rclpy  # noqa: F401
+except ModuleNotFoundError:
+    pytest.skip(allow_module_level=True)
+
+from ..ros import ROSLogging
+from .test_logging import TestLogging as _TestLogging
+
+
+class TestROSLogging(_TestLogging):
+    tc = ROSLogging

hw_device_mgr/mgr/mgr.py

@@ -47,8 +47,6 @@ def device_model_id(cls):
     command_out_defaults = dict(state_cmd=0, reset=0)
     command_out_data_types = dict(state_cmd="uint8", reset="bit")
 
-    update_rate = 10  # Hz
-
     ####################################################
     # Initialization
 
@@ -406,6 +404,7 @@ def fsm_finalize_command(self, e):
 
     def run(self):
         """Program main loop."""
+        update_period = 1.0 / self.mgr_config.get("update_rate", 10.0)
         while not self.shutdown:
             try:
                 self.read_update_write()
@@ -423,7 +422,7 @@ def run(self):
                 # the `sleep()` before the next update
                 self.fast_track = False
                 continue
-            self.rate.sleep()
+            time.sleep(update_period)
 
     def read_update_write(self):
         """

hw_device_mgr/mgr/tests/bogus_devices/mgr_config.yaml

@@ -1,5 +1,6 @@
 init_timeout: 30.0  # seconds
 goal_state_timeout: 5.0  # seconds
+update_rate:  20.0  # updates/second
 
 devices:
   # 6 DOF joints

hw_device_mgr/mgr_ros/tests/base_test_class.py

@@ -1,8 +1,13 @@
 from ...mgr.tests.base_test_class import BaseMgrTestClass
-from .bogus_devices.mgr import ROSHWDeviceMgrTest
 import yaml
 import pytest
 
+try:
+    import rclpy  # noqa: F401
+except ModuleNotFoundError:
+    pytest.skip(allow_module_level=True)
+from .bogus_devices.mgr import ROSHWDeviceMgrTest
+
 
 ###############################
 # Test class
@@ -75,7 +80,7 @@ def device_config_path(self, tmp_path, device_config, mock_rclpy):
         yield tmpfile
 
     def test_mock_rclpy_fixture(self, mock_rclpy):
-        from ..mgr import rclpy
+        from ..mgr import rclpy  # noqa:  F811
 
         node = rclpy.create_node("foo")
         assert node is self.node

package.xml

@@ -24,13 +24,16 @@
   <exec_depend>python-fysom</exec_depend>
   <exec_depend>python3-lxml</exec_depend>
 
+  <test_depend>ament_copyright</test_depend>
+  <test_depend>ament_flake8</test_depend>
+  <test_depend>ament_pep257</test_depend>
   <test_depend>python3-pytest</test_depend>
   <test_depend>python3-pytest-cov</test_depend>
   <test_depend>python3-pytest-mock</test_depend>
 
   <member_of_group>rosidl_interface_packages</member_of_group>
 
   <export>
-    <build_type>ament_cmake</build_type>
+    <build_type>ament_python</build_type>
   </export>
 </package>

setup.py

@@ -16,7 +16,11 @@
     "mgr_ros_hal",
 ]
 # Packages like hw_device_mgr.{pkg}.tests
-pkgs_t = ["devices"] + pkgs_bd
+pkgs_t = [
+    "devices",
+    "logging",
+    *pkgs_bd,
+]
 # Generate lists
 packages = (
     [