Add method to get the current logging directory

rclpy/CMakeLists.txt

@@ -17,6 +17,7 @@ endif()
 find_package(ament_cmake REQUIRED)
 find_package(ament_cmake_python REQUIRED)
 find_package(rcl REQUIRED)
+find_package(rcl_logging_interface REQUIRED)
 find_package(rcl_action REQUIRED)
 find_package(rcl_yaml_param_parser REQUIRED)
 find_package(rcutils REQUIRED)
@@ -142,6 +143,7 @@ add_library(
 configure_python_c_extension_library(rclpy_logging)
 ament_target_dependencies(rclpy_logging
   "rcutils"
+  "rcl_logging_interface"
 )
 
 # Signal handling library

rclpy/package.xml

@@ -22,6 +22,7 @@
 
   <depend>rmw_implementation</depend>
   <depend>rcl</depend>
+  <depend>rcl_logging_interface</depend>
   <depend>rcl_action</depend>
   <depend>rcl_yaml_param_parser</depend>
   <depend>unique_identifier_msgs</depend>

rclpy/rclpy/logging.py

@@ -14,6 +14,7 @@
 
 
 from enum import IntEnum
+from pathlib import Path
 
 from rclpy.impl.implementation_singleton import rclpy_logging_implementation as _rclpy_logging
 import rclpy.impl.rcutils_logger
@@ -70,3 +71,13 @@ def get_logger_effective_level(name):
 def get_logging_severity_from_string(log_severity):
     return LoggingSeverity(
         _rclpy_logging.rclpy_logging_severity_level_from_string(log_severity))
+
+
+def get_logging_directory() -> Path:
+    """
+    Return the current logging directory being used.
+
+    For more details, see .. c:function::
+    rcl_logging_ret_t rcl_logging_get_logging_directory(rcutils_allocator_t, char **)
+    """
+    return Path(_rclpy_logging.rclpy_logging_get_logging_directory())

rclpy/src/rclpy/_rclpy_logging.c

@@ -19,6 +19,8 @@
 #include <rcutils/logging.h>
 #include <rcutils/time.h>
 
+#include <rcl_logging_interface/rcl_logging_interface.h>
+
 /// Initialize the logging system.
 /**
  * \return None or
@@ -254,6 +256,27 @@ rclpy_get_fatal_logging_severity(PyObject * Py_UNUSED(self), PyObject * Py_UNUSE
   return PyLong_FromLongLong(RCUTILS_LOG_SEVERITY_FATAL);
 }
 
+/// Get the current logging directory from rcutils.
+/// \return Unicode UTF8 object containing the current logging directory.
+static PyObject *
+rclpy_logging_get_logging_directory(PyObject * Py_UNUSED(self), PyObject * Py_UNUSED(args))
+{
+  char * log_dir = NULL;
+  rcutils_allocator_t allocator = rcutils_get_default_allocator();
+  rcl_logging_ret_t ret = rcl_logging_get_logging_directory(allocator, &log_dir);
+  if (RCL_LOGGING_RET_OK != ret) {
+    PyErr_Format(
+      PyExc_RuntimeError,
+      "Failed to get current logging directory, error: \"%s\", return code: \"%d\"\n",
+      rcutils_get_error_string().str, ret);
+    rcutils_reset_error();
+    return NULL;
+  }
+  PyObject * py_log_dir = PyUnicode_DecodeFSDefault(log_dir);
+  allocator.deallocate(log_dir, allocator.state);
+  return py_log_dir;
+}
+
 /// Define the public methods of this module
 static PyMethodDef rclpy_logging_methods[] = {
   {
@@ -309,6 +332,10 @@ static PyMethodDef rclpy_logging_methods[] = {
     "rclpy_get_fatal_logging_severity", rclpy_get_fatal_logging_severity,
     METH_VARARGS, "Get log fatal severity level as int from rcutils"
   },
+  {
+    "rclpy_logging_get_logging_directory", rclpy_logging_get_logging_directory,
+    METH_VARARGS, "Get the current logging directory from rcutils"
+  },
 
   {NULL, NULL, 0, NULL}  /* sentinel */
 };

rclpy/test/test_logging.py

@@ -13,6 +13,8 @@
 # limitations under the License.
 
 import inspect
+import os
+from pathlib import Path
 import time
 import unittest
 
@@ -368,6 +370,15 @@ def test_nonexistent_logging_severity_from_string(self):
         with self.assertRaises(RuntimeError):
             rclpy.logging.get_logging_severity_from_string('non_existent_severity')
 
+    def test_get_logging_directory(self):
+        os.environ['HOME'] = '/fake_home_dir'
+        os.environ.pop('USERPROFILE', None)
+        os.environ.pop('ROS_LOG_DIR', None)
+        os.environ.pop('ROS_HOME', None)
+        log_dir = rclpy.logging.get_logging_directory()
+        assert isinstance(log_dir, Path)
+        assert log_dir == Path('/fake_home_dir') / '.ros' / 'log'
+
 
 if __name__ == '__main__':
     unittest.main()