pycallgraph.py

"""
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)