ModuleManager.py

Go to the documentation of this file.

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




import string
import sys,os
import glob

import OpenRTM_aist


CONFIG_EXT    = "manager.modules.config_ext"
CONFIG_PATH   = "manager.modules.config_path"
DETECT_MOD    = "manager.modules.detect_loadable"
MOD_LOADPTH   = "manager.modules.load_path"
INITFUNC_SFX  = "manager.modules.init_func_suffix"
INITFUNC_PFX  = "manager.modules.init_func_prefix"
ALLOW_ABSPATH = "manager.modules.abs_path_allowed"
ALLOW_URL     = "manager.modules.download_allowed"
MOD_DWNDIR    = "manager.modules.download_dir"
MOD_DELMOD    = "manager.modules.download_cleanup"
MOD_PRELOAD   = "manager.modules.preload"




class ModuleManager:
  """
  """



  
  def __init__(self, prop):
    self._properties = prop

    self._configPath = prop.getProperty(CONFIG_PATH).split(",")
    for i in range(len(self._configPath)):
      tmp = [self._configPath[i]]
      OpenRTM_aist.eraseHeadBlank(tmp)
      self._configPath[i] = tmp[0]
    self._loadPath = prop.getProperty(MOD_LOADPTH,"./").split(",")
    for i in range(len(self._loadPath)):
      tmp = [self._loadPath[i]]
      OpenRTM_aist.eraseHeadBlank(tmp)
      self._loadPath[i] = tmp[0]

    self._absoluteAllowed = OpenRTM_aist.toBool(prop.getProperty(ALLOW_ABSPATH),
                                                "yes", "no", False)

    self._downloadAllowed = OpenRTM_aist.toBool(prop.getProperty(ALLOW_URL),
                                                "yes", "no", False)

    self._initFuncSuffix = prop.getProperty(INITFUNC_SFX)
    self._initFuncPrefix = prop.getProperty(INITFUNC_PFX)
    self._modules = OpenRTM_aist.ObjectManager(self.DLLPred)
    self._rtcout = None
    self._mgr = OpenRTM_aist.Manager.instance()

  
  def __del__(self):
    self.unloadAll()


  
  class Error:
    def __init__(self, reason_):
      self.reason = reason_



  
  class NotFound:
    def __init__(self, name_):
      self.name = name_



  
  class FileNotFound(NotFound):
    def __init__(self, name_):
      ModuleManager.NotFound.__init__(self, name_)



  
  class ModuleNotFound(NotFound):
    def __init__(self, name_):
      ModuleManager.NotFound.__init__(self, name_)



  
  class SymbolNotFound(NotFound):
    def __init__(self, name_):
      ModuleManager.NotFound.__init__(self, name_)



  
  class NotAllowedOperation(Error):
    def __init__(self, reason_):
      ModuleManager.Error.__init__(self, reason_)
      ModuleManager.Error.__init__(self, reason_)



  
  class InvalidArguments(Error):
    def __init__(self, reason_):
      ModuleManager.Error.__init__(self, reason_)



  
  class InvalidOperation(Error):
    def __init__(self, reason_):
      ModuleManager.Error.__init__(self, reason_)



  
  def load(self, file_name, init_func=None):
    if not self._rtcout:
      self._rtcout = self._mgr.getLogbuf("ModuleManager")

    self._rtcout.RTC_TRACE("load(fname = %s)", file_name)
    if file_name == "":
      raise ModuleManager.InvalidArguments, "Invalid file name."

    if OpenRTM_aist.isURL(file_name):
      if not self._downloadAllowed:
        raise ModuleManager.NotAllowedOperation, "Downloading module is not allowed."
      else:
        raise ModuleManager.NotFound, "Not implemented."

    import_name = os.path.split(file_name)[-1]
    pathChanged=False
    file_path = None
    if OpenRTM_aist.isAbsolutePath(file_name):
      if not self._absoluteAllowed:
        raise ModuleManager.NotAllowedOperation, "Absolute path is not allowed"
      else:
        splitted_name = os.path.split(file_name)
        save_path = sys.path[:]
        sys.path.append(splitted_name[0])
        pathChanged = True
        import_name = splitted_name[-1]
        file_path = file_name

    else:
      file_path = self.findFile(file_name, self._loadPath)
      if not file_path:
        raise ModuleManager.InvalidArguments, "Invalid file name."

    if not self.fileExist(file_path):
      raise ModuleManager.FileNotFound, file_name

    if not pathChanged:
      splitted_name = os.path.split(file_path)
      sys.path.append(splitted_name[0])

    ext_pos = import_name.find(".py")
    if ext_pos > 0:
      import_name = import_name[:ext_pos]
    mo = __import__(str(import_name))

    if pathChanged:
      sys.path = save_path

    dll = self.DLLEntity(mo,OpenRTM_aist.Properties())
    dll.properties.setProperty("file_path",file_path)
    self._modules.registerObject(dll)


    if init_func is None:
      return file_name

    self.symbol(file_path,init_func)(self._mgr)

    return file_name


  
  def unload(self, file_name):
    dll = self._modules.find(file_name)
    if not dll:
      raise ModuleManager.NotFound, file_name

    self._modules.unregisterObject(file_name)
    return


  
  def unloadAll(self):
    dlls = self._modules.getObjects()
    
    for dll in dlls:
      ident = dll.properties.getProperty("file_path")
      self._modules.unregisterObject(ident)
    return


  
  def symbol(self, file_name, func_name):
    dll = self._modules.find(file_name)
    if not dll:
      raise ModuleManager.ModuleNotFound, file_name

    func = getattr(dll.dll,func_name,None)

    if not func:
      raise ModuleManager.SymbolNotFound, func_name
    
    return func


  
  def setLoadpath(self, load_path_list):
    self._loadPath = load_path_list
    return


  
  def getLoadPath(self):
    return self._loadPath


  
  def addLoadpath(self, load_path):
    for path in load_path:
      self._loadPath.append(path)
    return


  
  def getLoadedModules(self):
    dlls = self._modules.getObjects()
    modules = []
    for dll in dlls:
      modules.append(dll.properties)

    return modules


  def __getRtcProfile(self, fname):
    # file name with full path
    fullname  = fname
    # directory name
    dirname   = os.path.dirname(fname)
    sys.path.append(dirname)
    # basename
    basename  = os.path.basename(fname)
    # classname
    classname  = basename.split(".")[0].lower()

    # loaded profile = old profiles - new profiles
    # for old
    oldp = self._mgr.getFactoryProfiles()

    # for new
    comp_spec_name = classname+"_spec"

    try:
      imp_file = __import__(basename.split(".")[0])
    except:
      return None
    comp_spec = getattr(imp_file,comp_spec_name,None)
    if not comp_spec:
      return None
    newp = OpenRTM_aist.Properties(defaults_str=comp_spec)

    profs = []
    
    exists = False
    for i in range(len(oldp)):
      if    oldp[i].getProperty("implementation_id") == newp.getProperty("implementation_id") and \
            oldp[i].getProperty("type_name") == newp.getProperty("type_name") and \
            oldp[i].getProperty("description") == newp.getProperty("description") and \
            oldp[i].getProperty("version") == newp.getProperty("version"):
        exists = True
    if not exists:
      profs.append(newp)

        
    # loaded component profile have to be one
    if len(profs) == 0:
      return OpenRTM_aist.Properties()

    if len(profs) > 1:
      return None

    return profs[0]


  
  def getLoadableModules(self):
    # getting loadable module file path list.
    modules_ = []
    for path in self._loadPath:
      if path == "":
        continue

      flist = glob.glob(path + os.sep + '*.py')
      for file in flist:
        if file.find("__init__.py") == -1:
          modules_.append(file)
    
    props = []
    # getting module properties from loadable modules
    for mod_ in modules_:
      prop = self.__getRtcProfile(mod_)
      if prop:
        prop.setProperty("module_file_name",os.path.basename(mod_))
        prop.setProperty("module_file_path", mod_)
        props.append(prop)

    return props



  
  def allowAbsolutePath(self):
    self._absoluteAllowed = True


  
  def disallowAbsolutePath(self):
    self._absoluteAllowed = False


  
  def allowModuleDownload(self):
    self._downloadAllowed = True


  
  def disallowModuleDownload(self):
    self._downloadAllowed = False


  
  def findFile(self, fname, load_path):
    file_name = fname
    for path in load_path:
      if fname.find(".py") == -1:
        f = str(path) + os.sep + str(file_name)+".py"
      else:
        f = str(path)+ os.sep + str(file_name)
      if self.fileExist(f):
        return f
    return ""


  
  def fileExist(self, filename):
    fname = filename
    if fname.find(".py") == -1:
      fname = str(filename)+".py"

    if os.path.isfile(fname):
      return True

    return False



  
  def getInitFuncName(self, file_path):
    base_name = os.path.basename(file_path)
    return str(self._initFuncPrefix)+str(base_name)+str(self._initFuncSuffix)



  
  class DLL:
    def __init__(self, dll):
      self.dll = dll
      return


  class DLLEntity:
    def __init__(self,dll,prop):
      self.dll = dll
      self.properties = prop


  class DLLPred:
    def __init__(self, name=None, factory=None):
      self._filepath = name or factory

    def __call__(self, dll):
      return self._filepath == dll.properties.getProperty("file_path")