PK"LQJ.callgraph/__init__.py"""Decorators and Jupyter IPython magic to display a dynamic call graph.""" __version__ = '0.1.3' from .extension import load_ipython_extension from .decorator import decorator from .recorder import CallGraphRecorder PK"Lrܳcallgraph/decorator.pyfrom .recorder import CallGraphRecorder def decorator(fn=None, recorder=None, label_returns=False, graph_attrs=None): """A decorator that can be used to instrument a function.""" rec = recorder or CallGraphRecorder(label_returns=label_returns, graph_attrs=graph_attrs) def graphing_decorator(fn): wrapper = rec.wrap(fn) if not recorder: wrapper.__callgraph__ = rec.graph return wrapper return graphing_decorator(fn) if fn else graphing_decorator PKR"LW callgraph/extension.pyimport ast from IPython.core.magic import (Magics, line_magic, line_cell_magic, magics_class, needs_local_scope) from IPython.display import SVG, display from IPython.testing.skipdoctest import skip_doctest from .recorder import CallGraphRecorder from .instrumentor import CallGraphInstrumentor @magics_class class CallGraphMagics(Magics): @skip_doctest @line_cell_magic @needs_local_scope def callgraph(self, line, local_ns=None): """Display the dynamic call graph for a Python statement or expression. Usage: %callgraph nchoosek(5, 2) This magic instruments global functions that are named in the statement. On completion, the functions are restored to their original values. The magic knows about functions that are decorated with functools.lru_cache. Two calls with the same arguments to a cached function will display as the same node. For simplicity, "same" to the instrumentation just means same string representation. It also ignores the `maxsize` and `typed` arguments to lru_cache. Options: -r: reverse the graph (display arrows from callee to caller). Label the arrows with the return values. -h: display the graph “horizontally”, with function calls running from left to right. -w: max width of the graph Examples -------- ``` from functools import lru_cache @lru_cache() def nchoosek(n, k): if k == 0: return 1 if n == k: return 1 return nchoosek(n - 1, k - 1) + nchoosek(n - 1, k) %callgraph nchoosek(5, 2) %callgraph nchoosek(5, 2); nchoosek(5, 3) %callgraph -r nchoosek(5, 2) ``` """ # Some constants. filename = '' mode = 'exec' opts, stmt = self.parse_options(line, 'w:ehr', posix=False) # Parse the options. Create a call graph recorder with those options. options = {'graph_attrs': {}} if 'e' in opts: options['equal'] = True if 'r' in opts: options['label_returns'] = True if 'h' in opts: options['graph_attrs']['rankdir'] = 'LR' if 'w' in opts: options['graph_attrs']['size'] = '{},'.format(opts['w']) recorder = CallGraphRecorder(**options) # Parse the statement. Collect calls to instrument. tree = ast.parse(stmt, filename=filename, mode=mode) calls = [n for n in ast.walk(tree) if isinstance(n, ast.Call)] fns = [n.func.id for n in calls if isinstance(n.func, ast.Name)] # For now, only global variables (`a(x)`) are instrumented, not # attributes (`a.b(x)`). The following would collect attributes # ("qualified" function calls). # qfns = list(n.func for n in calls if isinstance(n.func, ast.Attribute)) with CallGraphInstrumentor(fns, recorder=recorder, local_ns=local_ns) as recorder: exec(compile(tree, filename=filename, mode=mode), local_ns) display(SVG(data=recorder.graph._repr_svg_())) def load_ipython_extension(ipython): "Register the IPython magic." ipython.register_magics(CallGraphMagics) PKf"L=$callgraph/instrumentor.pyfrom .recorder import CallGraphRecorder class CallGraphInstrumentor(object): def __init__(self, names, recorder=None, local_ns=None): self.recorder = recorder or CallGraphRecorder() self._names = names self._ns = local_ns or globals() def __enter__(self): ns = self._ns fns = {n: ns[n] for n in self._names if ns[n].__call__} for n, fn in fns.items(): ns[n] = self.recorder.wrap(fn) self._restore = fns return self.recorder def __exit__(self, type, value, traceback): ns = self._ns fns = self._restore for n, fn in fns.items(): ns[n] = fn PKۭ"Lbq q callgraph/recorder.pyimport functools from functools import wraps from itertools import starmap from graphviz import Digraph class CallGraphRecorder(object): def __init__(self, equal=False, label_returns=False, graph_attrs=None): self.graph = Digraph(format='svg', strict=True) if graph_attrs: self.graph.graph_attr.update(**graph_attrs) self._options = {'equal': equal, 'label_returns': label_returns} self._next_call_idx = 0 self._callers = [] def wrap(self, fn): "A decorator that wraps fn with instrumentation to record calls." @wraps(fn) def wrapper(*args, **kwargs): with self.record(fn, args, kwargs) as record_return: result = fn(*args, **kwargs) record_return(result) return result return wrapper def _record(self, caller_id, call_id, fn, args, kwargs, result): graph = self.graph label_returns = self._options['label_returns'] label = "{}({}{}{})".format(fn.__name__, ', '.join(map(repr, args)), ', ' if args and kwargs else '', ', '.join(starmap("{}={}".format, kwargs.items()))) if not (label_returns and caller_id): label += " ↦ {}".format(result) graph.node(call_id, label) if caller_id: if label_returns: graph.edge(caller_id, call_id, label=str(result), dir='back') else: graph.edge(caller_id, call_id) else: graph.node(call_id, penwidth='3') def record(self, fn, args, kwargs): """Usage: with recorder.record(fn, args, kwargs) as record_return: … record_return(result) """ return CallGraphCallRecorder(self, fn, args, kwargs) def _next_call_id(self, fn, args, kwargs): if self._options['equal'] or isinstance(fn, functools._lru_cache_wrapper): return '{}{}{}'.format(fn, args, kwargs) self._next_call_idx += 1 return str(self._next_call_idx) class CallGraphCallRecorder(object): __slots__ = ['_recorder', '_fn', '_args', '_kwargs'] def __init__(self, recorder, fn, args, kwargs): self._recorder = recorder self._fn, self._args, self._kwargs = fn, args, kwargs def __enter__(self): fn, args, kwargs = self._fn, self._args, self._kwargs recorder = self._recorder stack = recorder._callers caller_id = (stack[-1] if stack else None) call_id = recorder._next_call_id(fn, args, kwargs) stack.append(call_id) def record_return(result): recorder._record(caller_id, call_id, fn, args, kwargs, result) return record_return def __exit__(self, type, value, traceback): self._recorder._callers.pop() PK b"L0*88!callgraph-0.1.3.dist-info/LICENSEThe MIT License (MIT) Copyright (c) 2018 Oliver Steele 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. PK!HxQPcallgraph-0.1.3.dist-info/WHEEL1 0 RZq+D-Dv;_[*7Fp 8MRq%_:==ߘPT PK!HLeh"callgraph-0.1.3.dist-info/METADATAWn7}W PkNEmI;q-ɕ(~IܫdJYPaFǼT/I`ccYM@Scνcz@Z 3y)E9-JA o {7Klp.H>ֈ$ߦצcN28Y݀^

Tg.|( uÙ֫l(K)NsVQޝ @jG{=QͥP+ӕoQ' v!$j#/ļDGOi!-\7I|٥@t\l ;Dv'NsFUb#mdײA-B(ݑL] .n^O3j?c4 p3*ڈ(0PK"LQJ.callgraph/__init__.pyPK"Lrܳcallgraph/decorator.pyPKR"LW _callgraph/extension.pyPKf"L=$callgraph/instrumentor.pyPKۭ"Lbq q `callgraph/recorder.pyPK b"L0*88!callgraph-0.1.3.dist-info/LICENSEPK!HxQP{#callgraph-0.1.3.dist-info/WHEELPK!HLeh" $callgraph-0.1.3.dist-info/METADATAPK!H+;ȸ )callgraph-0.1.3.dist-info/RECORDPK +