PortProfileHelper.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- coding: euc-jp -*-
 
 
 
import threading
import OpenRTM_aist
import RTC
 
 
 
class PortProfileHelper:
  """
  """
 
  def __init__(self):
    self._mutex        = threading.RLock()
    self._name         = ""
    self._ifProfiles   = []
    self._portRef      = None
    self._connProfiles = []
    self._owner        = None
    self._properties   = []
    return
 
  
  def setPortProfile(self, profile):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    self._name         = profile.name
    self._ifProfiles   = profile.interfaces
    self._portRef      = profile.port_ref
    self._connProfiles = profile.connector_profiles
    self._owner        = profile.owner
    self._properties   = profile.properties
    return
 
  
  def getPortProfile(self):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    prof = RTC.PortProfile(self._name,
                           self._ifProfiles,
                           self._portRef,
                           self._connProfiles,
                           self._owner,
                           self._properties)
    return prof
 
 
  
  def setName(self, name):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    self._name = name
    return
 
  
  def getName(self):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    return self._name
 
 
  
  def appendPortInterfaceProfile(self, if_prof):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    self._ifProfiles.append(if_prof)
    return
 
  
  def getPortInterfaceProfiles(self):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    return self._ifProfiles
 
 
  
  def getPortInterfaceProfile(self, instance_name):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._ifProfiles,
                                            self.if_name(instance_name))
    if index < 0:
      return None
    else:
      return self._ifProfiles[index]
 
 
  
  def erasePortInterfaceProfile(self, instance_name):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._ifProfiles,
                                            self.if_name(instance_name))
    if index < 0:
      return
    else:
      del self._ifProfiles[index]
 
 
  
  def setPortRef(self, port):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    self._portRef = port
    return
 
 
  
  def getPortRef(self):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    return self._portRef
 
 
  
  def appendConnectorProfile(self, conn_profile):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    self._connProfiles.append(conn_profile)
    return
 
  
  def getConnectorProfiles(self):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    return self._connProfiles
 
 
  
  def getConnectorProfile(self, name):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._connProfiles,
                                            self.conn_name(name))
    if index < 0:
      return None
    else:
      return self._connProfiles[index]
 
    return None
 
 
  
  def getConnectorProfileById(self, id):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._connProfiles,
                                            self.conn_id(id))
    if index < 0:
      return None
    else:
      return self._connProfiles[index]
 
    return None
 
  
  def eraseConnectorProfile(self, name):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._connProfiles,
                                            self.conn_name(name))
    if index < 0:
      return
    else:
      del self._connProfiles[index]
 
    return
 
 
  
  def eraseConnectorProfileById(self, id):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._connProfiles,
                                            self.conn_id(id))
    if index < 0:
      return
    else:
      del self._connProfiles[index]
 
    return
 
  
  def setOwner(self, owner):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    self._owner = owner
    return
 
  
  def getOwner(self):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    return self._owner
 
 
  
  def setProperties(self, prop):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    self._properties = prop
    return
 
  
  def getProperties(self):
    guard = OpenRTM_aist.ScopedLock(self._mutex)
    return self._properties
 
 
  
  class if_name:
    def __init__(self, name):
      self._name = name
      return
 
    def __call__(self, prof):
      return str(self._name) == str(prof.instance_name)
 
 
  # Functor to find ConnectorProfile by name
  class conn_name:
    def __init__(self, name):
      self._name = name
      return
 
    def __call__(self, cprof):
      return str(self._name) == str(cprof.name)
 
    
  # Functor to find ConnectorProfile by id
  class conn_id:
    def __init__(self, id_):
      self._id = id_
      return
 
    def __call__(self, cprof):
      return str(self._id) == str(cprof.connector_id)