InPortCorbaCdrConsumer.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- coding: euc-jp -*-
 
 
 
import sys
from omniORB import any
from omniORB import CORBA
import OpenRTM_aist
import OpenRTM
 
 
class InPortCorbaCdrConsumer(OpenRTM_aist.InPortConsumer,OpenRTM_aist.CorbaConsumer):
  """
  """
 
  
  def __init__(self):
    OpenRTM_aist.CorbaConsumer.__init__(self)
    self._rtcout = OpenRTM_aist.Manager.instance().getLogbuf("InPortCorbaCdrConsumer")
    self._properties = None
    return
 
  
  def __del__(self, CorbaConsumer=OpenRTM_aist.CorbaConsumer):
    self._rtcout.RTC_PARANOID("~InPortCorbaCdrConsumer()")
    CorbaConsumer.__del__(self)
    return
 
  
  def init(self, prop):
    self._rtcout.RTC_TRACE("init()")
    self._properties = prop
    return
 
  
  def put(self, data):
    self._rtcout.RTC_PARANOID("put()")
 
    try:
      ref_ = self.getObject()
      if ref_:
        inportcdr = ref_._narrow(OpenRTM.InPortCdr)
        return self.convertReturnCode(inportcdr.put(data))
      return self.CONNECTION_LOST
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      return self.CONNECTION_LOST
        
    return self.UNKNOWN_ERROR
 
  
  def publishInterfaceProfile(self, properties):
    return
 
  
  def subscribeInterface(self, properties):
    self._rtcout.RTC_TRACE("subscribeInterface()")
    # self._rtcout.RTC_DEBUG_STR(OpenRTM_aist.NVUtil.toString(properties))
    
    # getting InPort's ref from IOR string
    if self.subscribeFromIor(properties):
      return True
    
    # getting InPort's ref from Object reference
    if self.subscribeFromRef(properties):
      return True
    
    return False
    
  
  def unsubscribeInterface(self, properties):
    self._rtcout.RTC_TRACE("unsubscribeInterface()")
    # self._rtcout.RTC_DEBUG_STR(OpenRTM_aist.NVUtil.toString(properties))
    
    if self.unsubscribeFromIor(properties):
      return
        
    self.unsubscribeFromRef(properties)
    return
 
  
  def subscribeFromIor(self, properties):
    self._rtcout.RTC_TRACE("subscribeFromIor()")
    
    index = OpenRTM_aist.NVUtil.find_index(properties,
                                           "dataport.corba_cdr.inport_ior")
    if index < 0:
      self._rtcout.RTC_ERROR("inport_ior not found")
      return False
    
    ior = ""
    try:
      ior = any.from_any(properties[index].value, keep_structs=True)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
 
    if not ior:
      self._rtcout.RTC_ERROR("inport_ior has no string")
      return False
    
    orb = OpenRTM_aist.Manager.instance().getORB()
    obj = orb.string_to_object(ior)
    
    if CORBA.is_nil(obj):
      self._rtcout.RTC_ERROR("invalid IOR string has been passed")
      return False
    
    if not self.setObject(obj):
      self._rtcout.RTC_WARN("Setting object to consumer failed.")
      return False
 
    return True
 
  
  def subscribeFromRef(self, properties):
    self._rtcout.RTC_TRACE("subscribeFromRef()")
    index = OpenRTM_aist.NVUtil.find_index(properties,
                                           "dataport.corba_cdr.inport_ref")
    if index < 0:
      self._rtcout.RTC_ERROR("inport_ref not found")
      return False
    
    obj = None
    try:
      obj = any.from_any(properties[index].value, keep_structs=True)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
    
    if not obj:
      self._rtcout.RTC_ERROR("prop[inport_ref] is not objref")
      return False
    
    if CORBA.is_nil(obj):
      self._rtcout.RTC_ERROR("prop[inport_ref] is not objref")
      return False
    
    if not self.setObject(obj):
      self._rtcout.RTC_ERROR("Setting object to consumer failed.")
      return False
 
    return True
 
  
  def unsubscribeFromIor(self, properties):
    self._rtcout.RTC_TRACE("unsubscribeFromIor()")
    index = OpenRTM_aist.NVUtil.find_index(properties,
                                           "dataport.corba_cdr.inport_ior")
    if index < 0:
      self._rtcout.RTC_ERROR("inport_ior not found")
      return False
    
    ior = ""
    try:
      ior = any.from_any(properties[index].value, keep_structs=True)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
 
    if not ior:
      self._rtcout.RTC_ERROR("prop[inport_ior] is not string")
      return False
    
    orb = OpenRTM_aist.Manager.instance().getORB()
    var = orb.string_to_object(ior)
    if not self._ptr()._is_equivalent(var):
      self._rtcout.RTC_ERROR("connector property inconsistency")
      return False
    
    self.releaseObject()
    return True
 
  
  def unsubscribeFromRef(self, properties):
    self._rtcout.RTC_TRACE("unsubscribeFromRef()")
    index = OpenRTM_aist.NVUtil.find_index(properties,
                                           "dataport.corba_cdr.inport_ref")
 
    if index < 0:
      return False
    
    obj = None
    try:
      obj = any.from_any(properties[index].value, keep_structs=True)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
    
    if not obj:
      return False
 
    if not self._ptr()._is_equivalent(obj):
      return False
    
    self.releaseObject()
    return True
 
  
  def convertReturnCode(self, ret):
    if ret == OpenRTM.PORT_OK:
      return self.PORT_OK
 
    elif ret == OpenRTM.PORT_ERROR:
      return self.PORT_ERROR
 
    elif ret == OpenRTM.BUFFER_FULL:
      return self.SEND_FULL
 
    elif ret == OpenRTM.BUFFER_TIMEOUT:
      return self.SEND_TIMEOUT
 
    elif ret == OpenRTM.UNKNOWN_ERROR:
      return self.UNKNOWN_ERROR
 
    else:
      return self.UNKNOWN_ERROR
 
 
def InPortCorbaCdrConsumerInit():
  factory = OpenRTM_aist.InPortConsumerFactory.instance()
  factory.addFactory("corba_cdr",
                     OpenRTM_aist.InPortCorbaCdrConsumer,
                     OpenRTM_aist.Delete)