diff --git a/rclpy/rclpy/impl/implementation_singleton.py b/rclpy/rclpy/impl/implementation_singleton.py index ba7f55fe8..cdbfc0b00 100644 --- a/rclpy/rclpy/impl/implementation_singleton.py +++ b/rclpy/rclpy/impl/implementation_singleton.py @@ -26,7 +26,19 @@ # ... """ + +from typing import List, Protocol, Sequence + from rpyutils import import_c_library package = 'rclpy' -rclpy_implementation = import_c_library('._rclpy_pybind11', package) +rclpy_implementation: 'rclpyHandle' = import_c_library('._rclpy_pybind11', package) + + +class rclpyHandle(Protocol): + + def rclpy_remove_ros_args(self, pycli_args: Sequence[str]) -> List[str]: + ... + + def rclpy_get_rmw_implementation_identifier(self) -> str: + ... diff --git a/rclpy/rclpy/utilities.py b/rclpy/rclpy/utilities.py index 1e6d501ca..654fd6efd 100644 --- a/rclpy/rclpy/utilities.py +++ b/rclpy/rclpy/utilities.py @@ -29,7 +29,7 @@ g_context_lock = threading.Lock() -def get_default_context(*, shutting_down=False) -> Context: +def get_default_context(*, shutting_down: bool = False) -> Context: """Return the global default context singleton.""" global g_context_lock with g_context_lock: @@ -70,7 +70,7 @@ def ok(*, context: Optional[Context] = None) -> bool: return context.ok() -def shutdown(*, context=None) -> None: +def shutdown(*, context: Optional[Context] = None) -> None: """ Shutdown the given ``Context``. @@ -82,7 +82,7 @@ def shutdown(*, context=None) -> None: context.shutdown() -def try_shutdown(*, context=None) -> None: +def try_shutdown(*, context: Optional[Context] = None) -> None: """ Shutdown the given ``Context`` if not already shutdown. @@ -131,8 +131,8 @@ def get_available_rmw_implementations() -> Set[str]: # filter by implementations in environment variable if provided rmw_implementations = os.environ.get('RMW_IMPLEMENTATIONS') if rmw_implementations: - rmw_implementations = rmw_implementations.split(os.pathsep) - missing_rmw_implementations = set(rmw_implementations) - \ + rmw_implementations_array = rmw_implementations.split(os.pathsep) + missing_rmw_implementations = set(rmw_implementations_array) - \ available_rmw_implementations if missing_rmw_implementations: # TODO(sloretz) function name suggets to me it would return available ones even