""" pycallgraph U{http://pycallgraph.slowchop.com/} Copyright Gerald Kaszuba 2007 This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA """ __version__ = '0.4.1' __author__ = 'Gerald Kaszuba' import inspect import sys import os import re import tempfile import time from distutils import sysconfig # Initialise module variables. # TODO Move these into settings trace_filter = None time_filter = None def colourize_node(calls, total_time): value = float(total_time * 2 + calls) / 3 return '%f %f %f' % (value / 2 + .5, value, 0.9) def colourize_edge(calls, total_time): value = float(total_time * 2 + calls) / 3 return '%f %f %f' % (value / 2 + .5, value, 0.7) def reset_settings(): global settings global graph_attributes global __version__ settings = { 'node_attributes': { 'label': r'%(func)s\ncalls: %(hits)i\ntotal time: %(total_time)f', 'color': '%(col)s', }, 'node_colour': colourize_node, 'edge_colour': colourize_edge, 'dont_exclude_anything': False, 'include_stdlib': True, } # TODO: Move this into settings graph_attributes = { 'graph': { 'fontname': 'Verdana', 'fontsize': 7, 'fontcolor': '0 0 0.5', 'label': r'Generated by Python Call Graph v%s\n' \ r'http://pycallgraph.slowchop.com' % __version__, }, 'node': { 'fontname': 'Verdana', 'fontsize': 7, 'color': '.5 0 .9', 'style': 'filled', 'shape': 'rect', }, } def reset_trace(): """Resets all collected statistics. This is run automatically by start_trace(reset=True) and when the module is loaded. """ global call_dict global call_stack global func_count global func_count_max global func_time global func_time_max global call_stack_timer call_dict = {} # current call stack call_stack = ['__main__'] # counters for each function func_count = {} func_count_max = 0 # accumative time per function func_time = {} func_time_max = 0 # keeps track of the start time of each call on the stack call_stack_timer = [] class PyCallGraphException(Exception): """Exception used for pycallgraph""" pass class GlobbingFilter(object): """Filter module names using a set of globs. Objects are matched against the exclude list first, then the include list. Anything that passes through without matching either, is excluded. """ def __init__(self, include=None, exclude=None, max_depth=None, min_depth=None): if include is None and exclude is None: include = ['*'] exclude = [] elif include is None: include = ['*'] elif exclude is None: exclude = [] self.include = include self.exclude = exclude self.max_depth = max_depth or 9999 self.min_depth = min_depth or 0 def __call__(self, stack, module_name=None, class_name=None, func_name=None, full_name=None): from fnmatch import fnmatch if len(stack) > self.max_depth: return False if len(stack) < self.min_depth: return False for pattern in self.exclude: if fnmatch(full_name, pattern): return False for pattern in self.include: if fnmatch(full_name, pattern): return True return False def is_module_stdlib(file_name): """Returns True if the file_name is in the lib directory.""" # TODO: Move these calls away from this function so it doesn't have to run # every time. lib_path = sysconfig.get_python_lib() path = os.path.split(lib_path) if path[1] == 'site-packages': lib_path = path[0] return file_name.lower().startswith(lib_path.lower()) def start_trace(reset=True, filter_func=None, time_filter_func=None): """Begins a trace. Setting reset to True will reset all previously recorded trace data. filter_func needs to point to a callable function that accepts the parameters (call_stack, module_name, class_name, func_name, full_name). Every call will be passed into this function and it is up to the function to decide if it should be included or not. Returning False means the call will be filtered out and not included in the call graph. """ global trace_filter global time_filter if reset: reset_trace() if filter_func: trace_filter = filter_func else: trace_filter = GlobbingFilter(exclude=['pycallgraph.*']) if time_filter_func: time_filter = time_filter_func else: time_filter = GlobbingFilter() sys.settrace(tracer) def stop_trace(): """Stops the currently running trace, if any.""" sys.settrace(None) def tracer(frame, event, arg): """This is an internal function that is called every time a call is made during a trace. It keeps track of relationships between calls. """ global func_count_max global func_count global trace_filter global time_filter global call_stack global func_time global func_time_max if event == 'call': keep = True code = frame.f_code # Stores all the parts of a human readable name of the current call. full_name_list = [] # Work out the module name module = inspect.getmodule(code) if module: module_name = module.__name__ module_path = module.__file__ if not settings['include_stdlib'] \ and is_module_stdlib(module_path): keep = False if module_name == '__main__': module_name = '' else: module_name = '' if module_name: full_name_list.append(module_name) # Work out the class name. try: class_name = frame.f_locals['self'].__class__.__name__ full_name_list.append(class_name) except (KeyError, AttributeError): class_name = '' # Work out the current function or method func_name = code.co_name if func_name == '?': func_name = '__main__' full_name_list.append(func_name) # Create a readable representation of the current call full_name = '.'.join(full_name_list) # Load the trace filter, if any. 'keep' determines if we should ignore # this call if keep and trace_filter: keep = trace_filter(call_stack, module_name, class_name, func_name, full_name) # Store the call information if keep: fr = call_stack[-1] if fr not in call_dict: call_dict[fr] = {} if full_name not in call_dict[fr]: call_dict[fr][full_name] = 0 call_dict[fr][full_name] += 1 if full_name not in func_count: func_count[full_name] = 0 func_count[full_name] += 1 if func_count[full_name] > func_count_max: func_count_max = func_count[full_name] call_stack.append(full_name) call_stack_timer.append(time.time()) else: call_stack.append('') call_stack_timer.append(None) if event == 'return': if call_stack: full_name = call_stack.pop(-1) t = call_stack_timer.pop(-1) if t and time_filter(stack=call_stack, full_name=full_name): if full_name not in func_time: func_time[full_name] = 0 call_time = (time.time() - t) func_time[full_name] += call_time if func_time[full_name] > func_time_max: func_time_max = func_time[full_name] return tracer def get_dot(stop=True): """Returns a string containing a DOT file. Setting stop to True will cause the trace to stop. """ global func_time_max def frac_calculation(func, count): global func_count_max global func_time global func_time_max calls_frac = float(count) / func_count_max try: total_time = func_time[func] except KeyError: total_time = 0 if func_time_max: total_time_frac = float(total_time) / func_time_max else: total_time_frac = 0 return calls_frac, total_time_frac, total_time if stop: stop_trace() ret = ['digraph G {', ] for comp, comp_attr in graph_attributes.items(): ret.append('%s [' % comp) for attr, val in comp_attr.items(): ret.append('%(attr)s = "%(val)s",' % locals()) ret.append('];') for func, hits in func_count.items(): calls_frac, total_time_frac, total_time = frac_calculation(func, hits) col = settings['node_colour'](calls_frac, total_time_frac) attribs = ['%s="%s"' % a for a in settings['node_attributes'].items()] node_str = '"%s" [%s];' % (func, ','.join(attribs)) ret.append(node_str % locals()) for fr_key, fr_val in call_dict.items(): if fr_key == '': continue for to_key, to_val in fr_val.items(): calls_frac, total_time_frac, totla_time = \ frac_calculation(to_key, to_val) col = settings['edge_colour'](calls_frac, total_time_frac) edge = '[ color = "%s" ]' % col ret.append('"%s"->"%s" %s' % (fr_key, to_key, edge)) ret.append('}') ret = '\n'.join(ret) return ret def save_dot(filename): """Generates a DOT file and writes it into filename.""" open(filename, 'w').write(get_dot()) def make_graph(filename, format=None, tool=None, stop=None): """This has been changed to make_dot_graph.""" raise PyCallGraphException( \ 'make_graph is depricated. Please use make_dot_graph') def make_dot_graph(filename, format='png', tool='dot', stop=True): """Creates a graph using a Graphviz tool that supports the dot language. It will output into a file specified by filename with the format specified. Setting stop to True will stop the current trace. """ if stop: stop_trace() # create a temporary file to be used for the dot data fd, tempname = tempfile.mkstemp() f = os.fdopen(fd, 'w') f.write(get_dot()) f.close() # normalize filename regex_user_expand = re.compile('\A~') if regex_user_expand.match(filename): filename = os.path.expanduser(filename) else: filename = os.path.expandvars(filename) # expand, just in case cmd = '%(tool)s -T%(format)s -o%(filename)s %(tempname)s' % locals() try: ret = os.system(cmd) if ret: raise PyCallGraphException( \ 'The command "%(cmd)s" failed with error ' \ 'code %(ret)i.' % locals()) finally: os.unlink(tempname) def simple_memoize(callable_object): """Simple memoization for functions without keyword arguments. This is useful for mapping code objects to module in this context. inspect.getmodule() requires a number of system calls, which may slow down the tracing considerably. Caching the mapping from code objects (there is *one* code object for each function, regardless of how many simultaneous activations records there are). In this context we can ignore keyword arguments, but a generic memoizer ought to take care of that as well. """ cache = dict() def wrapper(*rest): if rest not in cache: cache[rest] = callable_object(*rest) return cache[rest] return wrapper settings = {} graph_attributes = {} reset_settings() reset_trace() inspect.getmodule = simple_memoize(inspect.getmodule) # vim:set shiftwidth=4 tabstop=4 expandtab textwidth=79: