<s> """<STR_LIT>""" <EOL> import sys <EOL> import os <EOL> import imp <EOL> from . import log <EOL> from telex . plugin import NormalizePluginNameForModuleName <EOL> from telex . plugin . IPlugin import IPlugin <EOL> from telex . plugin . IPluginLocator import IPluginLocator <EOL> from telex . plugin . PluginFileLocator import PluginFileAnalyzerWithInfoFile <EOL> from telex . plugin . PluginFileLocator import PluginFileLocator <EOL> from telex . plugin import PLUGIN_NAME_FORBIDEN_STRING <EOL> from telex . plugin . PluginInfo import PluginInfo <EOL> class PluginManager ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> categories_filter = None , <EOL> directories_list = None , <EOL> plugin_info_ext = None , <EOL> plugin_locator = None ) : <EOL> """<STR_LIT>""" <EOL> if categories_filter is None : <EOL> categories_filter = { "<STR_LIT>" : IPlugin } <EOL> self . setCategoriesFilter ( categories_filter ) <EOL> plugin_locator = self . _locatorDecide ( plugin_info_ext , plugin_locator ) <EOL> self . setPluginLocator ( plugin_locator , directories_list ) <EOL> def _locatorDecide ( self , plugin_info_ext , plugin_locator ) : <EOL> """<STR_LIT>""" <EOL> specific_info_ext = plugin_info_ext is not None <EOL> specific_locator = plugin_locator is not None <EOL> if not specific_info_ext and not specific_locator : <EOL> res = PluginFileLocator ( ) <EOL> elif not specific_info_ext and specific_locator : <EOL> res = plugin_locator <EOL> elif not specific_locator and specific_info_ext : <EOL> res = PluginFileLocator ( ) <EOL> res . setAnalyzers ( [ PluginFileAnalyzerWithInfoFile ( "<STR_LIT>" , plugin_info_ext ) ] ) <EOL> elif specific_info_ext and specific_locator : <EOL> msg = ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> raise ValueError ( "<STR_LIT:U+0020>" . join ( msg ) % self . __class__ . __name__ ) <EOL> return res <EOL> def setCategoriesFilter ( self , categories_filter ) : <EOL> """<STR_LIT>""" <EOL> self . categories_interfaces = categories_filter . copy ( ) <EOL> self . category_mapping = { } <EOL> self . _category_file_mapping = { } <EOL> for categ in categories_filter : <EOL> self . category_mapping [ categ ] = [ ] <EOL> self . _category_file_mapping [ categ ] = [ ] <EOL> def setPluginPlaces ( self , directories_list ) : <EOL> """<STR_LIT>""" <EOL> self . getPluginLocator ( ) . setPluginPlaces ( directories_list ) <EOL> def updatePluginPlaces ( self , directories_list ) : <EOL> """<STR_LIT>""" <EOL> self . getPluginLocator ( ) . updatePluginPlaces ( directories_list ) <EOL> def setPluginInfoExtension ( self , ext ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . getPluginLocator ( ) . setPluginInfoExtension ( ext ) <EOL> except KeyError : <EOL> log . error ( "<STR_LIT>" ) <EOL> def setPluginInfoClass ( self , picls , strategies = None ) : <EOL> """<STR_LIT>""" <EOL> if strategies : <EOL> for name in strategies : <EOL> self . getPluginLocator ( ) . setPluginInfoClass ( picls , name ) <EOL> else : <EOL> self . getPluginLocator ( ) . setPluginInfoClass ( picls ) <EOL> def getPluginInfoClass ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . getPluginLocator ( ) . getPluginInfoClass ( ) <EOL> def setPluginLocator ( self , plugin_locator , dir_list = None , picls = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( plugin_locator , IPluginLocator ) : <EOL> self . _plugin_locator = plugin_locator <EOL> if dir_list is not None : <EOL> self . _plugin_locator . updatePluginPlaces ( dir_list ) <EOL> if picls is not None : <EOL> self . setPluginInfoClass ( picls ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" % plugin_locator ) <EOL> def getPluginLocator ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _plugin_locator <EOL> def _gatherCorePluginInfo ( self , directory , plugin_info_filename ) : <EOL> """<STR_LIT>""" <EOL> return self . getPluginLocator ( ) . gatherCorePluginInfo ( directory , plugin_info_filename ) <EOL> def _getPluginNameAndModuleFromStream ( self , infoFileObject , candidate_infofile = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return self . getPluginLocator ( ) . getPluginNameAndModuleFromStream ( infoFileObject , candidate_infofile ) <EOL> def getCategories ( self ) : <EOL> """<STR_LIT>""" <EOL> return list ( self . category_mapping . keys ( ) ) <EOL> def removePluginFromCategory ( self , plugin , category_name ) : <EOL> """<STR_LIT>""" <EOL> self . category_mapping [ category_name ] . remove ( plugin ) <EOL> def appendPluginToCategory ( self , plugin , category_name ) : <EOL> """<STR_LIT>""" <EOL> self . category_mapping [ category_name ] . append ( plugin ) <EOL> def getPluginsOfCategory ( self , category_name ) : <EOL> """<STR_LIT>""" <EOL> return self . category_mapping [ category_name ] [ : ] <EOL> def getAllPlugins ( self ) : <EOL> """<STR_LIT>""" <EOL> allPlugins = set ( ) <EOL> for pluginsOfOneCategory in self . category_mapping . values ( ) : <EOL> allPlugins . update ( pluginsOfOneCategory ) <EOL> return list ( allPlugins ) <EOL> def getPluginCandidates ( self ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> return self . _candidates [ : ] <EOL> def removePluginCandidate ( self , candidateTuple ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . _candidates . remove ( candidateTuple ) <EOL> def appendPluginCandidate ( self , candidateTuple ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . _candidates . append ( candidateTuple ) <EOL> def locatePlugins ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _candidates , npc = self . getPluginLocator ( ) . locatePlugins ( ) <EOL> def loadPlugins ( self , callback = None ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> processed_plugins = [ ] <EOL> for candidate_infofile , candidate_filepath , plugin_info in self . _candidates : <EOL> plugin_module_name_template = NormalizePluginNameForModuleName ( "<STR_LIT>" + plugin_info . name ) + "<STR_LIT>" <EOL> for plugin_name_suffix in range ( len ( sys . modules ) ) : <EOL> plugin_module_name = plugin_module_name_template % plugin_name_suffix <EOL> if plugin_module_name not in sys . modules : <EOL> break <EOL> if candidate_filepath . endswith ( "<STR_LIT>" ) : <EOL> candidate_filepath = candidate_filepath [ : - <NUM_LIT:3> ] <EOL> if callback is not None : <EOL> callback ( plugin_info ) <EOL> if "<STR_LIT>" in os . path . basename ( candidate_filepath ) : <EOL> candidate_filepath = os . path . dirname ( candidate_filepath ) <EOL> try : <EOL> if os . path . isdir ( candidate_filepath ) : <EOL> candidate_module = imp . load_module ( plugin_module_name , None , candidate_filepath , ( "<STR_LIT>" , "<STR_LIT:r>" , imp . PKG_DIRECTORY ) ) <EOL> else : <EOL> with open ( candidate_filepath + "<STR_LIT>" , "<STR_LIT:r>" ) as plugin_file : <EOL> candidate_module = imp . load_module ( plugin_module_name , plugin_file , candidate_filepath + "<STR_LIT>" , ( "<STR_LIT>" , "<STR_LIT:r>" , imp . PY_SOURCE ) ) <EOL> except Exception : <EOL> exc_info = sys . exc_info ( ) <EOL> log . error ( "<STR_LIT>" % candidate_filepath , exc_info = exc_info ) <EOL> plugin_info . error = exc_info <EOL> processed_plugins . append ( plugin_info ) <EOL> continue <EOL> processed_plugins . append ( plugin_info ) <EOL> if "<STR_LIT>" in os . path . basename ( candidate_filepath ) : <EOL> sys . path . remove ( plugin_info . path ) <EOL> for element in ( getattr ( candidate_module , name ) for name in dir ( candidate_module ) ) : <EOL> plugin_info_reference = None <EOL> for category_name in self . categories_interfaces : <EOL> try : <EOL> is_correct_subclass = issubclass ( element , self . categories_interfaces [ category_name ] ) <EOL> except Exception : <EOL> continue <EOL> if is_correct_subclass and element is not self . categories_interfaces [ category_name ] : <EOL> current_category = category_name <EOL> if candidate_infofile not in self . _category_file_mapping [ current_category ] : <EOL> if not plugin_info_reference : <EOL> try : <EOL> plugin_info . plugin_object = self . instanciateElement ( element ) <EOL> plugin_info_reference = plugin_info <EOL> except Exception : <EOL> exc_info = sys . exc_info ( ) <EOL> log . error ( "<STR_LIT>" % candidate_filepath , exc_info = exc_info ) <EOL> plugin_info . error = exc_info <EOL> break <EOL> plugin_info . categories . append ( current_category ) <EOL> self . category_mapping [ current_category ] . append ( plugin_info_reference ) <EOL> self . _category_file_mapping [ current_category ] . append ( candidate_infofile ) <EOL> delattr ( self , '<STR_LIT>' ) <EOL> return processed_plugins <EOL> def instanciateElement ( self , element ) : <EOL> """<STR_LIT>""" <EOL> return element ( ) <EOL> def collectPlugins ( self ) : <EOL> """<STR_LIT>""" <EOL> self . locatePlugins ( ) <EOL> self . loadPlugins ( ) <EOL> def getPluginByName ( self , name , category = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if category in self . category_mapping : <EOL> for item in self . category_mapping [ category ] : <EOL> if item . name == name : <EOL> return item <EOL> return None <EOL> def activatePluginByName ( self , name , category = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> pta_item = self . getPluginByName ( name , category ) <EOL> if pta_item is not None : <EOL> plugin_to_activate = pta_item . plugin_object <EOL> if plugin_to_activate is not None : <EOL> log . debug ( "<STR_LIT>" % ( category , name ) ) <EOL> plugin_to_activate . activate ( ) <EOL> return plugin_to_activate <EOL> return None <EOL> def deactivatePluginByName ( self , name , category = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if category in self . category_mapping : <EOL> plugin_to_deactivate = None <EOL> for item in self . category_mapping [ category ] : <EOL> if item . name == name : <EOL> plugin_to_deactivate = item . plugin_object <EOL> break <EOL> if plugin_to_deactivate is not None : <EOL> log . debug ( "<STR_LIT>" % ( category , name ) ) <EOL> plugin_to_deactivate . deactivate ( ) <EOL> return plugin_to_deactivate <EOL> return None <EOL> class PluginManagerSingleton ( object ) : <EOL> """<STR_LIT>""" <EOL> __instance = None <EOL> __decoration_chain = None <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . __instance is not None : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def setBehaviour ( self , list_of_pmd ) : <EOL> """<STR_LIT>""" <EOL> if self . __decoration_chain is None and self . __instance is None : <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . __decoration_chain = list_of_pmd <EOL> return True <EOL> else : <EOL> log . debug ( "<STR_LIT>" ) <EOL> return False <EOL> setBehaviour = classmethod ( setBehaviour ) <EOL> def get ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . __instance is None : <EOL> if self . __decoration_chain is not None : <EOL> pm = self . __decoration_chain [ <NUM_LIT:0> ] ( ) <EOL> for cls_item in self . __decoration_chain [ <NUM_LIT:1> : ] : <EOL> pm = cls_item ( decorated_manager = pm ) <EOL> self . __instance = pm <EOL> else : <EOL> self . __instance = PluginManager ( ) <EOL> log . debug ( "<STR_LIT>" ) <EOL> return self . __instance <EOL> get = classmethod ( get ) <EOL> from . PluginManagerDecorator import PluginManagerDecorator </s>
