Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_graphviz_data(self):
"""
Emit a graph representing the connections between the objects described
within the metadata list. The text representation can be transformed to
a graph with graphviz. Returns a string.
"""
s = []
header = '// Process this file with graphviz\n'
s.append( header)
s.append('digraph G {\n')
s.append(' node [shape=box];\n')
for md in self.metadata:
label = trunc(md.str, 48).replace('"', "'")
extra = ''
if md.type == 'instancemethod':
extra = ', color=red'
elif md.type == 'frame':
extra = ', color=orange'
s.append(' "X%s" [ label = "%s\\n%s" %s ];\n' % \
(hex(md.id)[1:], label, md.type, extra))
for e in self.edges:
extra = ''
if e.label == '__dict__':
extra = ',weight=100'
s.append(' X%s -> X%s [label="%s"%s];\n' % \
(hex(e.src)[1:], hex(e.dst)[1:], e.label, extra))
s.append('}\n')
return "".join(s)
def print_object(self, tobj):
"""
Print the gathered information of object `tobj` in human-readable format.
"""
if tobj.death:
self.stream.write('%-32s ( free ) %-35s\n' % (
trunc(tobj.name, 32, left=1), trunc(tobj.repr, 35)))
else:
self.stream.write('%-32s 0x%08x %-35s\n' % (
trunc(tobj.name, 32, left=1),
tobj.id,
trunc(tobj.repr, 35)
))
if tobj.trace:
self.stream.write(_format_trace(tobj.trace))
for (timestamp, size) in tobj.snapshots:
self.stream.write(' %-30s %s\n' % (
pp_timestamp(timestamp), pp(size.size)
))
self._print_refs(size.refs, size.size)
if tobj.death is not None:
self.stream.write(' %-30s finalize\n' % (
pp_timestamp(tobj.death),
))
fobj = self.stream
fobj.write('---- SUMMARY '+'-'*66+'\n')
for snapshot in self.snapshots:
self.annotate_snapshot(snapshot)
fobj.write('%-35s %11s %12s %12s %5s\n' % (
trunc(snapshot.desc, 35),
'active',
pp(snapshot.asizeof_total),
'average',
'pct'
))
for classname in classlist:
info = snapshot.classes.get(classname)
fobj.write(' %-33s %11d %12s %12s %4d%%\n' % (
trunc(classname, 33),
info['active'],
pp(info['sum']),
pp(info['avg']),
info['pct']
))
fobj.write('-'*79+'\n')
def print_object(self, tobj):
"""
Print the gathered information of object `tobj` in human-readable format.
"""
if tobj.death:
self.stream.write('%-32s ( free ) %-35s\n' % (
trunc(tobj.name, 32, left=1), trunc(tobj.repr, 35)))
else:
self.stream.write('%-32s 0x%08x %-35s\n' % (
trunc(tobj.name, 32, left=1),
tobj.id,
trunc(tobj.repr, 35)
))
if tobj.trace:
self.stream.write(_format_trace(tobj.trace))
for (timestamp, size) in tobj.snapshots:
self.stream.write(' %-30s %s\n' % (
pp_timestamp(timestamp), pp(size.size)
))
self._print_refs(size.refs, size.size)
if tobj.death is not None:
self.stream.write(' %-30s finalize\n' % (
pp_timestamp(tobj.death),
))
def print_object(self, tobj):
"""
Print the gathered information of object `tobj` in human-readable
format.
"""
if tobj.death:
self.stream.write('%-32s ( free ) %-35s\n' % (
trunc(tobj.name, 32, left=1), trunc(tobj.repr, 35)))
else:
self.stream.write('%-32s 0x%08x %-35s\n' % (
trunc(tobj.name, 32, left=1),
tobj.id,
trunc(tobj.repr, 35)
))
if tobj.trace:
self.stream.write(_format_trace(tobj.trace))
for (timestamp, size) in tobj.snapshots:
self.stream.write(' %-30s %s\n' % (
pp_timestamp(timestamp), pp(size.size)
))
self._print_refs(size.refs, size.size)
if tobj.death is not None:
self.stream.write(' %-30s finalize\n' % (
pp_timestamp(tobj.death),
def print_stats(self, stream=None):
"""
Log annotated garbage objects to console or file.
:param stream: open file, uses sys.stdout if not given
"""
if not stream: # pragma: no cover
stream = sys.stdout
self.metadata.sort(key=lambda x: -x.size)
stream.write('%-10s %8s %-12s %-46s\n' % ('id', 'size', 'type', 'representation'))
for g in self.metadata:
stream.write('0x%08x %8d %-12s %-46s\n' % (g.id, g.size, trunc(g.type, 12),
trunc(g.str, 46)))
stream.write('Garbage: %8d collected objects (%s in cycles): %12s\n' % \
(self.count, self.num_in_cycles, pp(self.total_size)))
def print_summary(self):
"""
Print per-class summary for each snapshot.
"""
# Emit class summaries for each snapshot
classlist = self.tracked_classes
fobj = self.stream
fobj.write('---- SUMMARY ' + '-' * 66 + '\n')
for snapshot in self.snapshots:
self.annotate_snapshot(snapshot)
fobj.write('%-35s %11s %12s %12s %5s\n' % (
trunc(snapshot.desc, 35),
'active',
pp(snapshot.asizeof_total),
'average',
'pct'
))
for classname in classlist:
info = snapshot.classes.get(classname)
fobj.write(' %-33s %11d %12s %12s %4d%%\n' % (
trunc(classname, 33),
info['active'],
pp(info['sum']),
pp(info['avg']),
info['pct']
))
fobj.write('-' * 79 + '\n')
def print_stats(self, stream=None):
"""
Log annotated garbage objects to console or file.
:param stream: open file, uses sys.stdout if not given
"""
if not stream: # pragma: no cover
stream = sys.stdout
self.metadata.sort(key=lambda x: -x.size)
stream.write('%-10s %8s %-12s %-46s\n' % ('id', 'size', 'type',
'representation'))
for g in self.metadata:
stream.write('0x%08x %8d %-12s %-46s\n' % (g.id, g.size,
trunc(g.type, 12),
trunc(g.str, 46)))
stream.write('Garbage: %8d collected objects (%s in cycles): %12s\n' %
(self.count, self.num_in_cycles, pp(self.total_size)))
def _print_refs(self, fobj, refs, total, level=1, minsize=0, minpct=0.1):
"""
Print individual referents recursively.
"""
lrefs = list(refs)
lrefs.sort(key=lambda x: x.size)
lrefs.reverse()
if level == 1:
fobj.write('\n')
for ref in lrefs:
if ref.size > minsize and (ref.size * 100.0 / total) > minpct:
data = dict(level=level,
name=trunc(str(ref.name), 128),
size=pp(ref.size),
pct=ref.size * 100.0 / total)
fobj.write(self.refrow % data)
self._print_refs(fobj, ref.refs, total, level=level + 1)
if level == 1:
fobj.write("<table></table>\n")