SdoConfiguration.py

Go to the documentation of this file.

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



import sys
import copy
import threading

import OpenRTM_aist

import SDOPackage, SDOPackage__POA



# SdoConfiguration with SeqEx 159120
# SdoConfiguration with SeqUtil 114504 114224



def toProperties(prop, conf):
  OpenRTM_aist.NVUtil.copyToProperties(prop, conf.configuration_data)



def toConfigurationSet(conf, prop):
  conf.description = prop.getProperty("description")
  conf.id = prop.getName()
  OpenRTM_aist.NVUtil.copyFromProperties(conf.configuration_data, prop)




class Configuration_impl(SDOPackage__POA.Configuration):
  """
  """

  
  def __init__(self, configAdmin, sdoServiceAdmin):
    """
     \var self._deviceProfile SDO DeviceProfile with mutex lock
    """
    self._deviceProfile = SDOPackage.DeviceProfile("","","","",[])
    self._dprofile_mutex = threading.RLock()

    """
     \var self._serviceProfiles SDO ServiceProfileList
    """
    self._serviceProfiles = []
    self._sprofile_mutex = threading.RLock()

    self._parameters = []
    self._params_mutex = threading.RLock()

    self._configsets = configAdmin
    self._config_mutex = threading.RLock()

    self._sdoservice = sdoServiceAdmin

    """
     \var self._organizations SDO OrganizationList
    """
    self._organizations = []
    self._org_mutex = threading.RLock()

    self._objref = self._this()
    self._rtcout = OpenRTM_aist.Manager.instance().getLogbuf("rtobject.sdo_config")


  #============================================================
  #
  # <<< CORBA interfaces >>>
  #
  #============================================================

  
  def set_device_profile(self, dProfile):
    self._rtcout.RTC_TRACE("set_device_profile()")
    if dProfile is None:
      raise SDOPackage.InvalidParameter("dProfile is empty.")

    try:
      guard = OpenRTM_aist.ScopedLock(self._dprofile_mutex)
      self._deviceProfile = dProfile
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Unknown Error")

    return True


  
  def add_service_profile(self, sProfile):
    self._rtcout.RTC_TRACE("add_service_profile()")
    if sProfile is None:
      raise SDOPackage.InvalidParameter("sProfile is empty.")

    try:
      return self._sdoservice.addSdoServiceConsumer(sProfile)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration.add_service_profile")

    return False


  
  def add_organization(self, org):
    self._rtcout.RTC_TRACE("add_organization()")
    if org is None:
      raise SDOPackage.InvalidParameter("org is empty.")

    try:
      OpenRTM_aist.CORBA_SeqUtil.push_back(self._organizations, org)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration.add_organization")

    return True


  
  def remove_service_profile(self, id_):
    self._rtcout.RTC_TRACE("remove_service_profile(%s)", id_)
    if id_ is None:
      raise SDOPackage.InvalidParameter("id is empty.")

    try:
      return self._sdoservice.removeSdoServiceConsumer(id_)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration.remove_service_profile")

    return False


  
  def remove_organization(self, organization_id):
    self._rtcout.RTC_TRACE("remove_organization(%s)", organization_id)
    if organization_id is None:
      raise SDOPackage.InvalidParameter("organization_id is empty.")

    try:
      guard = OpenRTM_aist.ScopedLock(self._org_mutex)
      OpenRTM_aist.CORBA_SeqUtil.erase_if(self._organizations,
                                          self.org_id(organization_id))
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration.remove_organization")

    return True


  
  def get_configuration_parameters(self):
    self._rtcout.RTC_TRACE("get_configuration_parameters()")
    try:
      guard = OpenRTM_aist.ScopedLock(self._params_mutex)
      param = copy.copy(self._parameters)
      return param
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration.get_configuration_parameters")

    return []


  
  def get_configuration_parameter_values(self):
    self._rtcout.RTC_TRACE("get_configuration_parameter_values()")
    guard = OpenRTM_aist.ScopedLock(self._config_mutex)
    nvlist = []
    return nvlist


  
  def get_configuration_parameter_value(self, name):
    self._rtcout.RTC_TRACE("get_configuration_parameter_value(%s)", name)
    if not name:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InvalidParameter("Name is empty.")

    return None


  
  def set_configuration_parameter(self, name, value):
    self._rtcout.RTC_TRACE("set_configuration_parameter(%s, value)", name)
    if name is None or value is None:
      raise SDOPackage.InvalidParameter("Name/Value is empty.")
    return True


  
  def get_configuration_sets(self):
    self._rtcout.RTC_TRACE("get_configuration_sets()")
    try:
      guard = OpenRTM_aist.ScopedLock(self._config_mutex)

      cf = self._configsets.getConfigurationSets()
      len_ = len(cf)

      config_sets = [SDOPackage.ConfigurationSet("","",[]) for i in range(len_)]
      for i in range(len_):
        toConfigurationSet(config_sets[i], cf[i])

      return config_sets

    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration.get_configuration_sets")

    return []

  
  def get_configuration_set(self, config_id):
    self._rtcout.RTC_TRACE("get_configuration_set(%s)", config_id)
    if not config_id:
      raise SDOPackage.InvalidParameter("ID is empty")

    guard = OpenRTM_aist.ScopedLock(self._config_mutex)

    try:
      if not self._configsets.haveConfig(config_id):
        self._rtcout.RTC_ERROR("No such ConfigurationSet")
        raise SDOPackage.InternalError("No such ConfigurationSet")
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Unknown exception")
      

    configset = self._configsets.getConfigurationSet(config_id)

    try:
      config = SDOPackage.ConfigurationSet("","",[])
      toConfigurationSet(config, configset)
      return config
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration::get_configuration_set()")

    return SDOPackage.ConfigurationSet("","",[])


  
  def set_configuration_set_values(self, configuration_set):
    self._rtcout.RTC_TRACE("set_configuration_set_values()")
    if not configuration_set or not configuration_set.id:
      raise SDOPackage.InvalidParameter("ID is empty.")

    try:
      conf = OpenRTM_aist.Properties(key=configuration_set.id)
      toProperties(conf, configuration_set)
      # ----------------------------------------------------------------------------
      # Because the format of port-name had been changed from <port_name> 
      # to <instance_name>.<port_name>, the following processing was added. 
      # (since r1648)

      if conf.findNode("exported_ports"):
        exported_ports = conf.getProperty("exported_ports").split(",")
        exported_ports_str = ""
        for i in range(len(exported_ports)):
          keyval = exported_ports[i].split(".")
          if len(keyval) > 2:
            exported_ports_str += keyval[0] + "." + keyval[-1]
          else:
            exported_ports_str += exported_ports[i]

          if i != (len(exported_ports)-1):
            exported_ports_str += ","
            
        conf.setProperty("exported_ports",exported_ports_str)
      #---------------------------------------------------------------------------
      return self._configsets.setConfigurationSetValues(conf)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration::set_configuration_set_values()")

    return True


  
  def get_active_configuration_set(self):
    self._rtcout.RTC_TRACE("get_active_configuration_set()")
    if not self._configsets.isActive():
      raise SDOPackage.NotAvailable("NotAvailable: Configuration.get_active_configuration_set()")

    try:
      guard = OpenRTM_aist.ScopedLock(self._config_mutex)
      config = SDOPackage.ConfigurationSet("","",[])
      toConfigurationSet(config, self._configsets.getActiveConfigurationSet())
      return config
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration.get_active_configuration_set()")

    return SDOPackage.ConfigurationSet("","",[])


  
  def add_configuration_set(self, configuration_set):
    self._rtcout.RTC_TRACE("add_configuration_set()")
    if configuration_set is None:
      raise SDOPackage.InvalidParameter("configuration_set is empty.")

    try:
      guard = OpenRTM_aist.ScopedLock(self._config_mutex)
      config_id = configuration_set.id
      config = OpenRTM_aist.Properties(key=config_id)
      toProperties(config, configuration_set)
      return self._configsets.addConfigurationSet(config)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration::add_configuration_set()")

    return True


  
  def remove_configuration_set(self, config_id):
    self._rtcout.RTC_TRACE("remove_configuration_set(%s)", config_id)
    if not config_id:
      raise SDOPackage.InvalidParameter("ID is empty.")
      
    try:
      guard = OpenRTM_aist.ScopedLock(self._config_mutex)
      return self._configsets.removeConfigurationSet(config_id)
    except:
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      raise SDOPackage.InternalError("Configuration.remove_configuration_set()")

    return False


  
  def activate_configuration_set(self, config_id):
    self._rtcout.RTC_TRACE("activate_configuration_set(%s)", config_id)
    if not config_id:
      raise SDOPackage.InvalidParameter("ID is empty.")

    if self._configsets.activateConfigurationSet(config_id):
      return True
    else:
      raise SDOPackage.InternalError("Configuration.activate_configuration_set()")

    return False


  #============================================================
  # end of CORBA interface definition
  #============================================================

  
  def getObjRef(self):
    return self._objref


  
  def getDeviceProfile(self):
    return self._deviceProfile


  
  def getServiceProfiles(self):
    return self._serviceProfiles


  
  def getServiceProfile(self, id):
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._serviceProfiles,
                                            self.service_id(id))

    if index < 0:
      return SDOPackage.ServiceProfile("","",[],None)

    return self._serviceProfiles[index]


  
  def getOrganizations(self):
    return self._organizations


  
  def getUUID(self):
    return OpenRTM_aist.uuid1()


  # functor for NVList
  
  class nv_name:
    def __init__(self, name_):
      self._name = str(name_)

    def __call__(self, nv):
      name_ = str(nv.name)
      return self._name == name_


  # functor for ServiceProfile
  
  class service_id:
    def __init__(self, id_):
      self._id = str(id_)

    def __call__(self, s):
      id_ = str(s.id)
      return self._id == id_


  # functor for Organization
  
  class org_id:
    def __init__(self, id_):
      self._id = str(id_)

    def __call__(self, o):
      id_ = str(o.get_organization_id())
      return self._id == id_

    
  # functor for ConfigurationSet
  
  class config_id:
    def __init__(self, id_):
      self._id = str(id_)

    def __call__(self, c):
      id_ = str(c.id)
      return self._id == id_