Source code for armarx_core.ice_manager

import os
import time

import logging
from typing import Any
from typing import TypeVar

from functools import lru_cache

import Ice
from Ice import NotRegisteredException

from IceGrid import RegistryPrx
from IceGrid import ObjectExistsException

from IceStorm import TopicManagerPrx
from IceStorm import NoSuchTopic
from IceStorm import AlreadySubscribed

from .config import get_ice_config_files
from .name_helper import get_ice_default_name

logger = logging.getLogger(__name__)

T = TypeVar("T")

[docs]def register_object( ice_object: Ice.Object, ice_object_name: str = None ) -> Ice.ObjectPrx: """ Register a local ice object under the given name :param ice_object: Local ice object instance :param ice_object_name: Name with which the object should be registered :return: Proxy to this object """ if not isinstance(ice_object, Ice.Object): logger.error("ice object is not an Ice.Object") raise ValueError("ice_object is not an Ice.Object") if not ice_object_name: ice_object_name = ice_object.__class__.__name__ logger.debug("Using class name %s to register the object", ice_object_name) adapter = freezer().communicator.createObjectAdapterWithEndpoints( ice_object_name, "tcp" ) ice_object_id = freezer().communicator.stringToIdentity(ice_object_name) adapter.add(ice_object, ice_object_id) adapter.activate() proxy = adapter.createProxy(ice_object_id) admin = get_admin() try:"adding new object %s", ice_object_name) admin.addObjectWithType(proxy, proxy.ice_id()) except ObjectExistsException:"updating new object %s", ice_object_name) admin.updateObject(proxy) return proxy
[docs]def get_topic(cls: T, topic_name: str = None) -> T: """ Retrieve a topic proxy casted to the first parameter :param cls: Type of the topic :param topic_name: Name of the topic :type topic_name: str :return: a casted topic proxy """ topic_name = topic_name or get_ice_default_name(cls) topic_manager = TopicManagerPrx.checkedCast( freezer().communicator.stringToProxy("IceStorm/TopicManager") ) topic = None try: topic = topic_manager.retrieve(topic_name) except NoSuchTopic: topic = topic_manager.create(topic_name)"Publishing to topic %s", topic_name) pub = topic.getPublisher().ice_oneway() return cls.uncheckedCast(pub)
[docs]def using_topic(proxy, topic_name: str = None): """ .. seealso:: :func:`register_object` :param proxy: the instance where the topic event should be called :param topic_name: the name of the topic to connect to :type topic_name: str """ topic_manager = TopicManagerPrx.checkedCast( freezer().communicator.stringToProxy("IceStorm/TopicManager") ) topic = None topic_name = topic_name or get_ice_default_name(proxy.__class__) try: topic = topic_manager.retrieve(topic_name) except NoSuchTopic: topic = topic_manager.create(topic_name) try: topic.subscribeAndGetPublisher(None, proxy) except AlreadySubscribed: topic.unsubscribe(proxy) topic.subscribeAndGetPublisher(None, proxy)"Subscribing to topic %s", topic_name) return topic
[docs]def wait_for_dependencies(proxy_names, timeout: int = 0): """ waits for a dependency list :param proxy_names: the proxy names to wait for :type proxy_names: list of str :returns: True if all dependencies are resolved :rtype: bool """ start_time = time.time() while not freezer().communicator.isShutdown(): if timeout and (start_time + timeout) < time.time(): logging.exception("Timeout while waiting for proxies %s", proxy_names) return False dependencies_resolved = True for proxy_name in proxy_names: try: proxy = freezer().communicator.stringToProxy(proxy_name) proxy.ice_ping() except NotRegisteredException: dependencies_resolved = False if dependencies_resolved: return True else: time.sleep(0.1)
[docs]def wait_for_proxy(cls, proxy_name: str = None, timeout: int = 0): """ waits for a proxy. :param cls: the class definition of an ArmarXComponent :param proxy_name: name of the proxy :param timeout: timeout in seconds to wait for the proxy. Zero means to wait forever :returns: the retrieved proxy :rtype: an instance of cls """ proxy = None proxy_name = proxy_name or get_ice_default_name(cls) start_time = time.time() while not freezer().communicator.isShutdown() and proxy is None: try: proxy = freezer().communicator.stringToProxy(proxy_name) proxy_cast = cls.checkedCast(proxy) return proxy_cast except NotRegisteredException: proxy = None if timeout and (start_time + timeout) < time.time(): logging.exception("Timeout while waiting for proxy %s", proxy_name) return None else: logger.debug("Waiting for proxy %s", proxy_name) time.sleep(0.1)
[docs]def get_proxy(cls: T, proxy_name: str = None) -> T: """ Connects to a proxy. :param cls: the class definition of an ArmarXComponent :param proxy_name: name of the proxy :type proxy_name: str :returns: the retrieved proxy :rtype: an instance of cls :raises: Ice::NotRegisteredException if the proxy is not available """ proxy_name = proxy_name or get_ice_default_name(cls) try: proxy = freezer().communicator.stringToProxy(proxy_name) return cls.checkedCast(proxy) except NotRegisteredException: logging.exception("Proxy %s does not exist", proxy_name)
def get_admin(): return freezer().registry.createAdminSession("user", "password").getAdmin() def is_connected(ice_node_name: str) -> bool: return get_admin().pingNode(ice_node_name)
[docs]def is_alive() -> bool: """ checks if shutdown has been invoked on the communicator. :returns: true if ice grid registry is alive """ return not freezer().communicator.isShutdown()
[docs]def wait_for_shutdown(): """ sleeps until the ice communicator receives a shutdown signal or the program receives a keyboard interrupt """ try: freezer().communicator.waitForShutdown() except KeyboardInterrupt: pass
[docs]@lru_cache(maxsize=1) def freezer(): """ """ return Freezer()
def test_connection(): if not is_connected("NodeMain"): logger.error("Ice is not running.") raise Exception("Ice is not running.") class Freezer: registry: RegistryPrx = None communicator = None def __init__(self): ice_config_files = get_ice_config_files() ice_communicator = Ice.initialize( ["--Ice.Config={}".format(",".join(ice_config_files))] ) ice_registry_proxy = ice_communicator.stringToProxy("IceGrid/Registry") ice_registry = RegistryPrx.checkedCast(ice_registry_proxy) self.communicator = ice_communicator self.registry = ice_registry import atexit atexit.register(self.__del__) @property def admin(self): return self.registry.createAdminSession("user", "password").getAdmin() def __enter__(self): return self.communicator def __exit__(self, exc_type, exc_val, exc_tb): pass def shutdown(self): if not self.communicator: return try: self.communicator.destroy() except Exception as ex: logger.error("Error while shutting down ice communicator") finally: self.communicator = None def __del__(self): self.shutdown()