CorbaPort.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- coding: euc-jp -*-



from omniORB import CORBA
from omniORB import any
import traceback
import sys

import OpenRTM_aist
import RTC




class CorbaPort(OpenRTM_aist.PortBase):
  """
  """

  
  def __init__(self, name):
    OpenRTM_aist.PortBase.__init__(self, name)
    self.addProperty("port.port_type", "CorbaPort")
    self._properties = OpenRTM_aist.Properties()
    self._providers = []
    self._consumers = []
    return


  def __del__(self, PortBase=OpenRTM_aist.PortBase):
    PortBase.__del__(self)


  
  def init(self, prop):
    self._rtcout.RTC_TRACE("init()")

    self._properties.mergeProperties(prop)

    num = [-1]
    if not OpenRTM_aist.stringTo([num], 
                                 self._properties.getProperty("connection_limit","-1")):
      self._rtcout.RTC_ERROR("invalid connection_limit value: %s", 
                             self._properties.getProperty("connection_limit"))

    self.setConnectionLimit(num[0])


  
  def registerProvider(self, instance_name, type_name, provider):
    self._rtcout.RTC_TRACE("registerProvider(instance=%s, type_name=%s)",
                           (instance_name, type_name))

    try:
      self._providers.append(self.CorbaProviderHolder(type_name,
                                                      instance_name,
                                                      provider))
    except:
      self._rtcout.RTC_ERROR("appending provider interface failed")
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      return False

    
    if not self.appendInterface(instance_name, type_name, RTC.PROVIDED):
      return False

    return True


  
  def registerConsumer(self, instance_name, type_name, consumer):
    self._rtcout.RTC_TRACE("registerConsumer()")

    if not self.appendInterface(instance_name, type_name, RTC.REQUIRED):
      return False
    
    self._consumers.append(self.CorbaConsumerHolder(type_name,
                                                    instance_name,
                                                    consumer,
                                                    self))
    return True


  
  def activateInterfaces(self):
    for provider in self._providers:
      provider.activate()

    return


  
  def deactivateInterfaces(self):
    for provider in self._providers:
      provider.deactivate()

    return



  
  def publishInterfaces(self, connector_profile):
    self._rtcout.RTC_TRACE("publishInterfaces()")

    returnvalue = self._publishInterfaces()

    if returnvalue != RTC.RTC_OK:
      return returnvalue

    properties = []
    for provider in self._providers:
      #------------------------------------------------------------
      # new version descriptor
      # <comp_iname>.port.<port_name>.provided.<type_name>.<instance_name>
      newdesc = self._profile.name[:len(self._ownerInstanceName)] + \
          ".port" +  self._profile.name[len(self._ownerInstanceName):]
      newdesc += ".provided." + provider.descriptor()

      properties.append(OpenRTM_aist.NVUtil.newNV(newdesc, provider.ior()))

      #------------------------------------------------------------
      # old version descriptor
      # port.<type_name>.<instance_name>
      olddesc = "port." + provider.descriptor()
      properties.append(OpenRTM_aist.NVUtil.newNV(olddesc, provider.ior()))

    OpenRTM_aist.CORBA_SeqUtil.push_back_list(connector_profile.properties, properties)
    
    return RTC.RTC_OK


  
  def subscribeInterfaces(self, connector_profile):
    self._rtcout.RTC_TRACE("subscribeInterfaces()")
    nv = connector_profile.properties

    strict = False # default is "best_effort"
    index = OpenRTM_aist.NVUtil.find_index(nv, "port.connection.strictness")
    if index >=  0:
      strictness = str(any.from_any(nv[index].value, keep_structs=True))
      if "best_effort" == strictness:
        strict = False
      elif "strict" == strictness:
        strict = True

      self._rtcout.RTC_DEBUG("Connetion strictness is: %s",strictness)

    for consumer in self._consumers:
      ior = []
      if self.findProvider(nv, consumer, ior) and len(ior) > 0:
        self.setObject(ior[0], consumer)
        continue

      ior = []
      if self.findProviderOld(nv, consumer, ior) and len(ior) > 0:
        self.setObject(ior[0], consumer)
        continue

      # never come here without error
      # if strict connection option is set, error is returned.
      if strict:
        self._rtcout.RTC_ERROR("subscribeInterfaces() failed.")
        return RTC.RTC_ERROR

    self._rtcout.RTC_TRACE("subscribeInterfaces() successfully finished.")

    return RTC.RTC_OK


  
  def unsubscribeInterfaces(self, connector_profile):
    self._rtcout.RTC_TRACE("unsubscribeInterfaces()")
    nv = connector_profile.properties

    for consumer in self._consumers:
      ior = []
      if self.findProvider(nv, consumer, ior) and len(ior) > 0:
        self._rtcout.RTC_DEBUG("Correspoinding consumer found.")
        self.releaseObject(ior[0], consumer)
        continue

      ior = []
      if self.findProviderOld(nv, consumer, ior) and len(ior) > 0:
        self._rtcout.RTC_DEBUG("Correspoinding consumer found.")
        self.releaseObject(ior[0], consumer)
        continue

    return


  
  def findProvider(self, nv, cons, iorstr):
    # new consumer interface descriptor
    newdesc = self._profile.name[:len(self._ownerInstanceName)] + \
        ".port" +  self._profile.name[len(self._ownerInstanceName):]
    newdesc += ".required." + cons.descriptor()

    # find a NameValue of the consumer
    cons_index = OpenRTM_aist.NVUtil.find_index(nv, newdesc)
    if cons_index < 0:
      return False

    provider = str(any.from_any(nv[cons_index].value, keep_structs=True))
    if not provider:
      self._rtcout.RTC_WARN("Cannot extract Provider interface descriptor")
      return False

    # find a NameValue of the provider
    prov_index = OpenRTM_aist.NVUtil.find_index(nv, provider)
    if prov_index < 0:
      return False

    ior_ = str(any.from_any(nv[prov_index].value, keep_structs=True))
    if not ior_:
      self._rtcout.RTC_WARN("Cannot extract Provider IOR string")
      return False
 
    if isinstance(iorstr, list):
      iorstr.append(ior_)

    self._rtcout.RTC_ERROR("interface matched with new descriptor: %s", newdesc)

    return True


  
  def findProviderOld(self, nv, cons, iorstr):
    # old consumer interface descriptor
    olddesc = "port." + cons.descriptor()

    # find a NameValue of the provider same as olddesc
    index = OpenRTM_aist.NVUtil.find_index(nv, olddesc)
    if index < 0:
      return False

    ior_ = str(any.from_any(nv[index].value, keep_structs=True))
    if not ior_:
      self._rtcout.RTC_WARN("Cannot extract Provider IOR string")
      return False

    if isinstance(iorstr, list):
      iorstr.append(ior_)

    self._rtcout.RTC_ERROR("interface matched with old descriptor: %s", olddesc)

    return True


  
  def setObject(self, ior, cons):
    # if ior string is "null" or "nil", ignore it.
    if "null" == ior:
      return True

    if "nil"  == ior:
      return True

    # IOR should be started by "IOR:"
    if "IOR:" != ior[:4]:
      return False

    # set IOR to the consumer
    if not cons.setObject(ior):
      self._rtcout.RTC_ERROR("Cannot narrow reference")
      return False

    self._rtcout.RTC_TRACE("setObject() done")
    return True

  
  def releaseObject(self, ior, cons):
    if ior == cons.getIor():
      cons.releaseObject()
      self._rtcout.RTC_DEBUG("Consumer %s released.", cons.descriptor())
      return True

    self._rtcout.RTC_WARN("IORs between Consumer and Connector are different.")
    return False

  
  class CorbaProviderHolder:
    # CorbaProviderHolder(const char* type_name,
    #                     const char* instance_name,
    #                     PortableServer::RefCountServantBase* servant)
    def __init__(self, type_name, instance_name, servant):
      self._typeName = type_name
      self._instanceName = instance_name
      self._servant = servant
      _mgr = OpenRTM_aist.Manager.instance()
      self._oid = _mgr.getPOA().servant_to_id(self._servant)

      obj = _mgr.getPOA().id_to_reference(self._oid)
      self._ior = _mgr.getORB().object_to_string(obj)
      self.deactivate()
      return

    def __del__(self):
      self.deactivate()
      
    # std::string instanceName() { return m_instanceName; }
    def instanceName(self):
      return self._instanceName

    # std::string typeName() { return m_typeName; }
    def typeName(self):
      return self._typeName

    # std::string ior() { return m_ior; }
    def ior(self):
      return self._ior

    # std::string descriptor() { return m_typeName + "." + m_instanceName; }
    def descriptor(self):
      return self._typeName + "." + self._instanceName

    # void activate()
    def activate(self):
      try:
        OpenRTM_aist.Manager.instance().getPOA().activate_object_with_id(self._oid, self._servant)
      except:
        print OpenRTM_aist.Logger.print_exception()
      return

    # void deactivate()
    def deactivate(self):
      try:
        OpenRTM_aist.Manager.instance().getPOA().deactivate_object(self._oid)
      except:
        print OpenRTM_aist.Logger.print_exception()
      return
    

  
  class CorbaConsumerHolder:
    # CorbaConsumerHolder(const char* type_name,
    #                     const char* instance_name,
    #                     CorbaConsumerBase* consumer,
    #                     string& owner)
    def __init__(self, type_name, instance_name, consumer, owner):
      self._typeName = type_name
      self._instanceName = instance_name
      self._consumer = consumer
      self._owner = owner
      self._ior = ""
      return

    # std::string instanceName() { return m_instanceName; }
    def instanceName(self):
      return self._instanceName

    # std::string typeName() { return m_typeName; }
    def typeName(self):
      return self._typeName

    # std::string descriptor() { return m_typeName + "." + m_instanceName; }
    def descriptor(self):
      return self._typeName + "." + self._instanceName

    
    def setObject(self, ior):
      self._ior = ior
      orb = OpenRTM_aist.Manager.instance().getORB()
      obj = orb.string_to_object(ior)
      if CORBA.is_nil(obj):
        return False

      return self._consumer.setObject(obj)

    
    def releaseObject(self):
      self._consumer.releaseObject()
      return

    # const std::string& getIor()
    def getIor(self):
      return self._ior


  
  class subscribe:
    def __init__(self, cons):
      self._cons = cons
      self._len  = len(cons)

    def __call__(self, nv):
      for i in range(self._len):
        name_ = nv.name
        if self._cons[i].descriptor() == name_:
          try:
            obj = any.from_any(nv.value, keep_structs=True)
            self._cons[i].setObject(obj)
          except:
            print OpenRTM_aist.Logger.print_exception()



  
  class unsubscribe:
    def __init__(self, cons):
      self._cons = cons
      self._len  = len(cons)

    def __call__(self, nv):
      for i in range(self._len):
        name_ = nv.name
        if self._cons[i].descriptor() == name_:
          self._cons[i].releaseObject()
          return

        # for 0.4.x
        if "port."+self._cons[i].descriptor() == name_:
          self._cons[i].releaseObject()