Project CMSSW displayed by LXR CMSSW_15_0_X_2024-12-03-2300/​FWCore/​ParameterSet/​python/​TreeCrawler.py	Source navigation Diff markup Identifier search General search
	Version: CMSSW_15_0_X_2024-12-03-2300 CMSSW_15_0_X_2024-12-02-2300 CMSSW_15_0_X_2024-12-01-2300 CMSSW_15_0_X_2024-11-28-2300 CMSSW_15_0_X_2024-11-27-2300 Revert   Change

File indexing completed on 2024-12-01 23:40:20

 # This CMS code is based on previous work done by Toby Dickenson, as indiciated below
 #
 # for questions: Benedikt.Hegner@cern.ch
 
 # Copyright 2004 Toby Dickenson
 #
 # Permission is hereby granted, free of charge, to any person obtaining
 # a copy of this software and associated documentation files (the
 # "Software"), to deal in the Software without restriction, including
 # without limitation the rights to use, copy, modify, merge, publish,
 # distribute, sublicense, and/or sell copies of the Software, and to
 # permit persons to whom the Software is furnished to do so, subject
 # to the following conditions:
 #
 # The above copyright notice and this permission notice shall be included
 # in all copies or substantial portions of the Software.
 #
 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
 # IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
 # CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
 # TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
 # SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 from builtins import range
 import sys, os, inspect, copy, struct, dis, importlib
 import modulefinder
 
 def packageNameFromFilename(name:str) -> str:
     return ".".join(name.replace("python/","").replace(".py","").split("/")[-3:])
 
 
 class Color:
   """ANSI escape display sequences"""
   info          = "\033[1;34m"
   hilight       = "\033[31m"
   alternate     = "\033[32m"
   extra         = "\033[33m"
   backlight     = "\033[43m"
   underline     = "\033[4m"
   lessemphasis  = "\033[30m"
   deemphasis    = "\033[1;30m"
   none          = "\033[0m"
 
 _stack = []
 
 class SearchHit:
     pass
 
 class Package(object):
     def __init__(self,name,top=False):
         self.name = name
         self.dependencies = []
         self.searched = False
         self.stack = []
         if top:
             self.module = None
         else:    
             self.module = __import__(name,[],[],"*")
     def dump(self,level:int):
         indent = "  " * level
         print(indent, "+", Color.info, self.name, Color.none)
         # sort dependencies alphabetically
         self.dependencies.sort(key = lambda x: x.name)
         for package in self.dependencies:
             package.dump(level+1)
     def search(self,pattern,result):
         """ recursive search for pattern in source files"""
         # first start searching in the package itself / do this only once
         if self.module:
             for number, line in enumerate(inspect.getsource(self.module).splitlines()):
                 if pattern in line:
                      filename = packageNameFromFilename(inspect.getsourcefile(self.module))
                      if not self.searched:
                          # save the hit, so we can add later stacks to it
                          self.hit = SearchHit()
                          self.hit.number = number
                          self.hit.filename = filename
                          self.hit.line = line
                          self.hit.stacks = list()
                          result.append(self.hit)
                      self.hit.stacks.append(copy.copy(_stack)) 
         # then go on with dependencies
         _stack.append(self.name)
         for package in self.dependencies:
             package.search(pattern,result)
         _stack.pop() 
         self.searched = True    
 
 
 class mymf(modulefinder.ModuleFinder):
     def __init__(self,*args,**kwargs):
         self._depgraph = {}
         self._types = {}
         self._last_caller = None
         #TODO - replace by environment variables CMSSW_BASE and CMSSW_RELEASE_BASE (*and* do it only if the global one is not empty like for IB areas)  
         self._localarea = os.path.expandvars('$CMSSW_BASE')
         self._globalarea = os.path.expandvars('$CMSSW_RELEASE_BASE')
         modulefinder.ModuleFinder.__init__(self,*args,**kwargs)
     def import_hook(self, name, caller=None, fromlist=None, level=-1):
         old_last_caller = self._last_caller
         try:
             self._last_caller = caller
             return modulefinder.ModuleFinder.import_hook(self,name,caller,fromlist, level=level)  
         finally:
             self._last_caller = old_last_caller
 
     def import_module(self,partnam,fqname,parent):
                               
         if partnam in ("os","unittest"):
             r = None
         else:
             r = modulefinder.ModuleFinder.import_module(self,partnam,fqname,parent)
             # since the modulefinder is not able to look into the global area when coming from the local area, we force a second try   
             if parent and not r and self._localarea != '' and self._globalarea != '':
                  parent.__file__ = parent.__file__.replace(self._localarea,self._globalarea)
                  parent.__path__[0] = parent.__path__[0].replace(self._localarea,self._globalarea)
             r = modulefinder.ModuleFinder.import_module(self,partnam,fqname,parent)
                                                          
         if r is not None:
             self._depgraph.setdefault(self._last_caller.__name__,{})[r.__name__] = 1
         return r
     def load_module(self, fqname, fp, pathname, aux_info):
         (suffix, mode, type) = aux_info
         r = modulefinder.ModuleFinder.load_module(self, fqname, fp, pathname, (suffix, mode, type))
         if r is not None:
             self._types[r.__name__] = type
         return r
 
     def scan_opcodes_25(self, co, unpack = struct.unpack):
         """
         This is basically just the default opcode scanner from ModuleFinder, but extended to also
         look for "process.load(<module>)' commands. Since the Process object might not necassarily
         be called "process", it scans for a call to a "load" method with a single parameter on
         *any* object. If one is found it checks if the parameter is a string that refers to a valid
         python module in the local or global area. If it does, the scanner assumes this was a call
         to a Process object and yields the module name.
         It's not possible to scan first for Process object declarations to get the name of the
         objects since often (e.g. for customisation functions) the object is passed to a function
         in a different file.
 
         The ModuleFinder.scan_opcodes_25 implementation this is based was taken from
         https://hg.python.org/cpython/file/2.7/Lib/modulefinder.py#l364
         """
         # Scan the code, and yield 'interesting' opcode combinations
         # Python 2.5 version (has absolute and relative imports)
         code = co.co_code
         names = co.co_names
         consts = co.co_consts
         LOAD_CONST = modulefinder.LOAD_CONST
         IMPORT_NAME = modulefinder.IMPORT_NAME
         STORE_OPS = modulefinder.STORE_OPS
         HAVE_ARGUMENT = modulefinder.HAVE_ARGUMENT
         LOAD_ATTR = chr(dis.opname.index('LOAD_ATTR'))
         LOAD_NAME = chr(dis.opname.index('LOAD_NAME'))
         CALL_FUNCTION = chr(dis.opname.index('CALL_FUNCTION'))
         LOAD_LOAD_AND_IMPORT = LOAD_CONST + LOAD_CONST + IMPORT_NAME
         
         try :
             indexOfLoadConst = names.index("load") # This might throw a ValueError
             # These are the opcodes required to access the "load" attribute. This might
             # not even be a function, but I check for that later.
             loadMethodOpcodes = LOAD_ATTR+struct.pack('<H',indexOfLoadConst)
         except ValueError :
             # doesn't look like "load" is used anywhere in this file
             loadMethodOpcodes=None
 
         while code:
             c = code[0]
             
             # Check to see if this is a call to a "load" method
             if loadMethodOpcodes!=None and len(code)>=9 : # Need at least 9 codes for the full call
                 if code[:3]==loadMethodOpcodes :
                     # The attribute "load" is being accessed, need to make sure this is a function call.
                     # I'll look ahead and see if the CALL_FUNCTION code is used - this could be in a different
                     # place depending on the number of arguments, but I'm only interested in methods with a
                     # single argument so I know exactly where CALL_FUNCTION should be.
                     if code[6]==CALL_FUNCTION :
                         # I know this is calling a method called "load" with one argument. I need
                         # to find out what the argument is. Note that I still don't know if this is
                         # on a cms.Process object.
                         indexInTable=unpack('<H',code[4:6])[0]
                         if code[3]==LOAD_CONST :
                             # The argument is a constant, so retrieve that from the table
                             loadMethodArgument=consts[indexInTable]
                             # I know a load method with one argument has been called on *something*, but I don't
                             # know if it was a cms.Process object. All I can do is check to see if the argument is
                             # a string, and if so if it refers to a python file in the user or global areas.
                             try :
                                 loadMethodArgument = loadMethodArgument.replace("/",".")
                                 # I can only use imp.find_module on submodules (i.e. each bit between a "."), so try
                                 # that on each submodule in turn using the previously found filename. Note that I have
                                 # to try this twice, because if the first pass traverses into a package in the local
                                 # area but the subpackage has not been checked out it will report that the subpackage
                                 # doesn't exist, even though it is available in the global area.
                                 try :
                                     parentFilename=[self._localarea+"/python"]
                                     for subModule in loadMethodArgument.split(".") :
                                         moduleInfo=importlib.machinery.PathFinder.find_spec( subModule, parentFilename )
                                         parentFilename=[moduleInfo.origin]
                                     # If control got this far without raising an exception, then it must be a valid python module
                                     yield "import", (None, loadMethodArgument)
                                 except ImportError :
                                     # Didn't work in the local area, try in the global area.
                                     parentFilename=[self._globalarea+"/python"]
                                     for subModule in loadMethodArgument.split(".") :
                                         moduleInfo=importlib.machinery.PathFinder.find_spec( subModule, parentFilename )
                                         parentFilename=[moduleInfo.origin]
                                     # If control got this far without raising an exception, then it must be a valid python module
                                     yield "import", (None, loadMethodArgument)
                             except Exception as error:
                                 # Either there was an import error (not a python module) or there was a string
                                 # manipulaton error (argument not a string). Assume this wasn't a call on a
                                 # cms.Process object and move on silently.
                                 pass
                         
                         elif code[3]==LOAD_NAME :
                             # The argument is a variable. I can get the name of the variable quite easily but
                             # not the value, unless I execute all of the opcodes. Not sure what to do here,
                             # guess I'll just print a warning so that the user knows?
                             print("Unable to determine the value of variable '"+names[indexInTable]+"' to see if it is a proces.load(...) statement in file "+co.co_filename)
                         
                         code=code[9:]
                         continue
 
             if c in STORE_OPS:
                 oparg, = unpack('<H', code[1:3])
                 yield "store", (names[oparg],)
                 code = code[3:]
                 continue
             if code[:9:3] == LOAD_LOAD_AND_IMPORT:
                 oparg_1, oparg_2, oparg_3 = unpack('<xHxHxH', code[:9])
                 level = consts[oparg_1]
                 if level == -1: # normal import
                     yield "import", (consts[oparg_2], names[oparg_3])
                 elif level == 0: # absolute import
                     yield "absolute_import", (consts[oparg_2], names[oparg_3])
                 else: # relative import
                     yield "relative_import", (level, consts[oparg_2], names[oparg_3])
                 code = code[9:]
                 continue
             if c >= HAVE_ARGUMENT:
                 code = code[3:]
             else:
                 code = code[1:]
 
 def removeRecursiveLoops( node, verbose=False, currentStack=None ) :
     if currentStack is None : currentStack=[]
     try :
         duplicateIndex=currentStack.index( node ) # If there isn't a recursive loop this will raise a ValueError
         if verbose :
             print("Removing recursive loop in:")
             for index in range(duplicateIndex,len(currentStack)) :
                 print("   ",currentStack[index].name,"-->")
             print("   ",node.name)
         currentStack[-1].dependencies.remove(node)
     except ValueError:
         # No recursive loop found, so continue traversing the tree
         currentStack.append( node )
         for subnode in node.dependencies :
             removeRecursiveLoops( subnode, verbose, currentStack[:] )
 
 def transformIntoGraph(depgraph,toplevel):
     packageDict = {}
     # create the top level config
     packageDict[toplevel] = Package(toplevel, top = True) 
 
     # create package objects
     for key, value in depgraph.items():
         if key.count(".") == 2 and key != toplevel: 
             packageDict[key] = Package(key)
         for name in value.keys():
             if name.count(".") == 2: packageDict[name] = Package(name)
     # now create dependencies
     for key, value in depgraph.items():
         if key.count(".") == 2 or key == toplevel:
             package = packageDict[key]
             package.dependencies = [packageDict[name] for name in value.keys() if name.count(".") == 2]
 
     removeRecursiveLoops( packageDict[toplevel] )
     # find and return the top level config
     return packageDict[toplevel]
 
 
 def getDependenciesFromPythonFile(filename:str,toplevelname,path):
     modulefinder = mymf(path)
     modulefinder.run_script(filename)
     globalDependencyDict = modulefinder._depgraph
     globalDependencyDict[toplevelname] = globalDependencyDict["__main__"] 
     return globalDependencyDict
 
 
 def getImportTree(filename:str,path):
     toplevelname = packageNameFromFilename(filename)
     # get dependencies from given file
     globalDependencyDict = getDependenciesFromPythonFile(filename,toplevelname,path)
         
     # transform this flat structure in a dependency tree
     dependencyGraph = transformIntoGraph(globalDependencyDict,toplevelname)
     return dependencyGraph                                               

This page was automatically generated by the 2.2.1 LXR engine. The LXR team