Add Navigator client

armarx_navigation/client.py

+import Ice
+
+import typing as ty
+
+from armarx_core import ice_manager
+from armarx_navigation.ice_interface import NavigatorInterfacePrx
+from armarx_core.reconnecting_proxy_client import ReconnectingProxyClient
+
+class Navigator:
+
+    DEFAULT_ICE_OBJECT_NAME = "navigator"
+
+    def __init__(
+            self,
+            proxy: NavigatorInterfacePrx,
+    ):
+        self.proxy = proxy
+
+    @classmethod
+    def wait_for_navigator(
+            cls,
+            name=DEFAULT_ICE_OBJECT_NAME,
+    ) -> "Navigator":
+        return cls(proxy=ice_manager.wait_for_proxy(NavigatorInterfacePrx, name))
+
+    @classmethod
+    def get_navigator(
+            cls,
+            name=DEFAULT_ICE_OBJECT_NAME,
+    ) -> ty.Optional["Navigator"]:
+        try:
+            return cls(proxy=ice_manager.get_proxy(NavigatorInterfacePrx, name))
+        except Ice.ConnectionRefusedException:
+            return None
+
+    def stop_all(self):
+        self.proxy.stopAll()
+
+
+class ReconnectingNavigator(ReconnectingProxyClient):
+
+    def __init__(
+            self,
+            name=Navigator.DEFAULT_ICE_OBJECT_NAME,
+    ):
+        super().__init__()
+        self.name = name
+
+    def get(self, wait=True, log=None) -> ty.Optional[Navigator]:
+        return self._get(wait=wait, log=log)
+
+    def _get_client(self, log=None) -> ty.Optional[Navigator]:
+        if log is not None:
+            log.info(f"Waiting for navigator '{self.name}'.")
+
+        client = Navigator.wait_for_navigator(name=self.name)
+
+        if log is not None:
+            log.info(f"Connected to navigator '{self.name}'.")
+
+        return client
+
+    def _wait_for_client(self, log=None) -> Navigator:
+        if log is not None:
+            log.info(f"Getting navigator '{self.name}' ...")
+
+        client = Navigator.get_navigator(name=self.name)
+
+        if log is not None:
+            if client is None:
+                log.info(f"Failed to connect to navigator '{self.name}'.")
+            else:
+                log.info(f"Connected to navigator '{self.name}'.")
+
+        return client

armarx_navigation/ice_interface.py

+from armarx_core import slice_loader
+
+slice_loader.load_armarx_slice("armarx_navigation", "../client/ice/NavigatorInterface.ice")
+
+from armarx.navigation.client import NavigatorInterface, NavigatorInterfacePrx