ConfigAdmin.py

Go to the documentation of this file.

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





import copy
import OpenRTM_aist


class OnUpdateCallback:
  def __init__(self):
    pass


  def __call__(self, config_set):
    pass



class OnUpdateParamCallback:
  def __init__(self):
    pass


  def __call__(self, config_set, config_param):
    pass



class OnSetConfigurationSetCallback:
  def __init__(self):
    pass


  def __call__(self, config_set):
    pass



class OnAddConfigurationAddCallback:
  def __init__(self):
    pass


  def __call__(self, config_set):
    pass



class OnRemoveConfigurationSetCallback:
  def __init__(self):
    pass


  def __call__(self, config_set):
    pass



class OnActivateSetCallback:
  def __init__(self):
    pass


  def __call__(self, config_id):
    pass




class Config:
  """
  """

  
  def __init__(self, name, var, def_val, trans=None):
    self.name = name
    self.default_value = def_val
    self._var = var
    if trans:
      self._trans = trans
    else:
      self._trans = OpenRTM_aist.stringTo


  
  def update(self, val):
    if self._trans(self._var, val):
      return True
    self._trans(self._var, self._default_value)
    return False




class ConfigAdmin:
  """
  """

  
  def __init__(self, configsets):
    self._configsets = configsets
    self._activeId   = "default"
    self._active     = True
    self._changed    = False
    self._params     = []
    self._emptyconf  = OpenRTM_aist.Properties()
    self._newConfig  = []
    self._listeners  = OpenRTM_aist.ConfigurationListeners()

  
  def __del__(self):
    del self._params


  
  def bindParameter(self, param_name, var, def_val, trans=None):
    if trans is None:
      trans = OpenRTM_aist.stringTo
    
    if self.isExist(param_name):
      return False

    if not trans(var, def_val):
      return False
    
    self._params.append(Config(param_name, var, def_val, trans))
    return True


  
  def update(self, config_set=None, config_param=None):
    # update(const char* config_set)
    if config_set and config_param is None:
      if self._configsets.hasKey(config_set) is None:
        return
      prop = self._configsets.getNode(config_set)
      for i in range(len(self._params)):
        if prop.hasKey(self._params[i].name):
          self._params[i].update(prop.getProperty(self._params[i].name))
          self.onUpdate(config_set)

    # update(const char* config_set, const char* config_param)
    if config_set and config_param:
      key = config_set
      key = key+"."+config_param
      for conf in self._params:
        if conf.name == config_param:
          conf.update(self._configsets.getProperty(key))
          self.onUpdateParam(config_set, config_param)
          return

    # update()
    if config_set is None and config_param is None:
      if self._changed and self._active:
        self.update(self._activeId)
        self._changed = False
      return


  
  def isExist(self, param_name):
    if not self._params:
      return False
    
    for conf in self._params:
      if conf.name == param_name:
        return True

    return False


  
  def isChanged(self):
    return self._changed


  
  def getActiveId(self):
    return self._activeId


  
  def haveConfig(self, config_id):
    if self._configsets.hasKey(config_id) is None:
      return False
    else:
      return True


  
  def isActive(self):
    return self._active


  
  def getConfigurationSets(self):
    return self._configsets.getLeaf()


  
  def getConfigurationSet(self, config_id):
    prop = self._configsets.getNode(config_id)
    if prop is None:
      return self._emptyconf
    return prop


  
  def setConfigurationSetValues(self, config_set):
    if config_set.getName() == "" or config_set.getName() is None:
      return False

    if not self._configsets.hasKey(config_set.getName()):
      return False

    p = self._configsets.getNode(config_set.getName())
    if p is None:
      return False

    p.mergeProperties(config_set)
    self._changed = True
    self._active  = False
    self.onSetConfigurationSet(config_set)
    return True


  
  def getActiveConfigurationSet(self):
    p = self._configsets.getNode(self._activeId)
    if p is None:
      return self._emptyconf

    return p


  
  def addConfigurationSet(self, configset):
    if self._configsets.hasKey(configset.getName()):
      return False
    node = configset.getName()

    # Create node
    self._configsets.createNode(node)

    p = self._configsets.getNode(node)
    if p is None:
      return False

    p.mergeProperties(configset)
    self._newConfig.append(node)

    self._changed = True
    self._active  = False
    self.onAddConfigurationSet(configset)
    return True


  
  def removeConfigurationSet(self, config_id):
    if config_id == "default":
      return False
    if self._activeId == config_id:
      return False

    find_flg = False
    # removeable config-set is only config-sets newly added
    for (idx,conf) in enumerate(self._newConfig):
      if conf == config_id:
        find_flg = True
        break


    if not find_flg:
      return False

    p = self._configsets.getNode(config_id)
    if p:
      p.getRoot().removeNode(config_id)
      del p

    del self._newConfig[idx]

    self._changed = True
    self._active  = False
    self.onRemoveConfigurationSet(config_id)
    return True


  
  def activateConfigurationSet(self, config_id):
    if config_id is None:
      return False

    # '_<conf_name>' is special configuration set name
    if config_id[0] == '_':
      return False

    if not self._configsets.hasKey(config_id):
      return False
    self._activeId = config_id
    self._active   = True
    self._changed  = True
    self.onActivateSet(config_id)
    return True


  #------------------------------------------------------------
  # obsolete functions
  #

  
  def setOnUpdate(self, cb):
    print "setOnUpdate function is obsolete."
    print "Use addConfigurationSetNameListener instead."
    self._listeners.configsetname_[OpenRTM_aist.ConfigurationSetNameListenerType.ON_UPDATE_CONFIG_SET].addListener(cb, False)
    return


  
  def setOnUpdateParam(self, cb):
    print "setOnUpdateParam function is obsolete."
    print "Use addConfigurationParamListener instead."
    self._listeners.configparam_[OpenRTM_aist.ConfigurationParamListenerType.ON_UPDATE_CONFIG_PARAM].addListener(cb, False)
    return


  
  def setOnSetConfigurationSet(self, cb):
    print "setOnSetConfigurationSet function is obsolete."
    print "Use addConfigurationSetListener instead."
    self._listeners.configset_[OpenRTM_aist.ConfigurationSetListenerType.ON_SET_CONFIG_SET].addListener(cb, False)
    return


  
  def setOnAddConfigurationSet(self, cb):
    print "setOnAddConfigurationSet function is obsolete."
    print "Use addConfigurationSetListener instead."
    self._listeners.configset_[OpenRTM_aist.ConfigurationSetListenerType.ON_ADD_CONFIG_SET].addListener(cb, False)
    return


  
  def setOnRemoveConfigurationSet(self, cb):
    print "setOnRemoveConfigurationSet function is obsolete."
    print "Use addConfigurationSetNameListener instead."
    self._listeners.configsetname_[OpenRTM_aist.ConfigurationSetNameListenerType.ON_REMOVE_CONFIG_SET].addListener(cb, False)
    return


  
  def setOnActivateSet(self, cb):
    print "setOnActivateSet function is obsolete."
    print "Use addConfigurationSetNameListener instead."
    self._listeners.configsetname_[OpenRTM_aist.ConfigurationSetNameListenerType.ON_ACTIVATE_CONFIG_SET].addListener(cb, False)
    return

  #
  # end of obsolete functions
  #------------------------------------------------------------

  
  def addConfigurationParamListener(self, type, listener, autoclean = True):
    self._listeners.configparam_[type].addListener(listener, autoclean)
    return


  
  def removeConfigurationParamListener(self, type, listener):
    self._listeners.configparam_[type].removeListener(listener)
    return
    

  
  def addConfigurationSetListener(self, type, listener, autoclean = True):
    self._listeners.configset_[type].addListener(listener, autoclean)
    return


  
  def removeConfigurationSetListener(self, type, listener):
    self._listeners.configset_[type].removeListener(listener)
    return
    

  
  def addConfigurationSetNameListener(self, type, listener, autoclean = True):
    self._listeners.configsetname_[type].addListener(listener, autoclean)
    return


  
  def removeConfigurationSetNameListener(self, type, listener):
    self._listeners.configsetname_[type].removeListener(listener)
    return
    

  
  def onUpdate(self, config_set):
    self._listeners.configsetname_[OpenRTM_aist.ConfigurationSetNameListenerType.ON_UPDATE_CONFIG_SET].notify(config_set)
    return


  
  def onUpdateParam(self, config_set, config_param):
    self._listeners.configparam_[OpenRTM_aist.ConfigurationParamListenerType.ON_UPDATE_CONFIG_PARAM].notify(config_set,
                                                                                                            config_param)
    return


  
  def onSetConfigurationSet(self, config_set):
    self._listeners.configset_[OpenRTM_aist.ConfigurationSetListenerType.ON_SET_CONFIG_SET].notify(config_set)
    return


  
  def onAddConfigurationSet(self, config_set):
    self._listeners.configset_[OpenRTM_aist.ConfigurationSetListenerType.ON_ADD_CONFIG_SET].notify(config_set)
    return


  
  def onRemoveConfigurationSet(self, config_id):
    self._listeners.configsetname_[OpenRTM_aist.ConfigurationSetNameListenerType.ON_REMOVE_CONFIG_SET].notify(config_id)
    return


  
  def onActivateSet(self, config_id):
    self._listeners.configsetname_[OpenRTM_aist.ConfigurationSetNameListenerType.ON_ACTIVATE_CONFIG_SET].notify(config_id)
    return


  class find_conf:
    def __init__(self, name):
      self._name = name
      return

    def __call__(self, conf):
      if conf is None or conf is 0:
        return False

      return self._name == conf.name