Source code for direct.showbase.ObjectReport

# pylint: skip-file

"""
>>> from direct.showbase import ObjectReport

>>> o=ObjectReport.ObjectReport('baseline')
>>> run()
...

>>> o2=ObjectReport.ObjectReport('')
>>> o.diff(o2)
"""

__all__ = ['ExclusiveObjectPool', 'ObjectReport']

from direct.directnotify.DirectNotifyGlobal import directNotify
from direct.showbase.DirectObject import DirectObject
from direct.showbase.ObjectPool import ObjectPool
from direct.showbase.GarbageReport import GarbageReport
from direct.showbase.PythonUtil import makeList, Sync, SerialNumGen
import gc
import sys
import builtins


[docs]class ExclusiveObjectPool(DirectObject): # ObjectPool specialization that excludes particular objects # IDs of objects to globally exclude from reporting _ExclObjs = [] _ExclObjIds = {} _SyncMaster = Sync('ExclusiveObjectPool.ExcludedObjectList') _SerialNumGen = SerialNumGen()
[docs] @classmethod def addExclObjs(cls, *objs): for obj in makeList(objs): if id(obj) not in cls._ExclObjIds: cls._ExclObjs.append(obj) cls._ExclObjIds.setdefault(id(obj), 0) cls._ExclObjIds[id(obj)] += 1 cls._SyncMaster.change()
[docs] @classmethod def removeExclObjs(cls, *objs): for obj in makeList(objs): assert id(obj) in cls._ExclObjIds cls._ExclObjIds[id(obj)] -= 1 if cls._ExclObjIds[id(obj)] == 0: del cls._ExclObjIds[id(obj)] cls._ExclObjs.remove(obj) cls._SyncMaster.change()
[docs] def __init__(self, objects): self._objects = list(objects) self._postFilterObjs = [] self._sync = Sync('%s-%s' % (self.__class__.__name__, self._SerialNumGen.next()), self._SyncMaster) self._sync.invalidate() ExclusiveObjectPool.addExclObjs(self._objects, self._postFilterObjs, self._sync)
[docs] def destroy(self): self.ignoreAll() ExclusiveObjectPool.removeExclObjs(self._objects, self._postFilterObjs, self._sync) del self._objects del self._postFilterObjs del self._sync
def _resync(self): if self._sync.isSynced(self._SyncMaster): return if hasattr(self, '_filteredPool'): ExclusiveObjectPool.removeExclObjs(*self._filteredPool._getInternalObjs()) ExclusiveObjectPool.removeExclObjs(self._filteredPool) del self._filteredPool del self._postFilterObjs[:] for obj in self._objects: if id(obj) not in ExclusiveObjectPool._ExclObjIds: self._postFilterObjs.append(obj) self._filteredPool = ExclusiveObjectPool(self._postFilterObjs) ExclusiveObjectPool.addExclObjs(self._filteredPool) ExclusiveObjectPool.addExclObjs(*self._filteredPool._getInternalObjs()) self._sync.sync(self._SyncMaster)
[docs] def getObjsOfType(self, type): self._resync() return self._filteredPool.getObjsOfType(type)
[docs] def printObjsOfType(self, type): self._resync() return self._filteredPool.printObjsOfType(type)
[docs] def diff(self, other): self._resync() return self._filteredPool.diff(other._filteredPool)
[docs] def typeFreqStr(self): self._resync() return self._filteredPool.typeFreqStr()
def __len__(self): self._resync() return len(self._filteredPool)
[docs]class ObjectReport: """report on every Python object in the current process""" notify = directNotify.newCategory('ObjectReport')
[docs] def __init__(self, name, log=True): gr = GarbageReport('ObjectReport\'s GarbageReport: %s' % name, log=log) gr.destroy() del gr self._name = name self._pool = ObjectPool(self._getObjectList()) #ExclusiveObjectPool.addExclObjs(self, self._pool, self._name) if log: self.notify.info('===== ObjectReport: \'%s\' =====\n%s' % (self._name, self.typeFreqStr()))
[docs] def destroy(self): #ExclusiveObjectPool.removeExclObjs(self, self._pool, self._name) self._pool.destroy() del self._pool del self._name
[docs] def typeFreqStr(self): return self._pool.typeFreqStr()
[docs] def diff(self, other): return self._pool.diff(other._pool)
[docs] def getObjectPool(self): return self._pool
def _getObjectList(self): if hasattr(sys, 'getobjects'): return sys.getobjects(0) else: gc.collect() # grab gc's object list gc_objects = gc.get_objects() # use get_referents to find everything else objects = gc_objects objects.append(builtins.__dict__) nextObjList = gc_objects found = set() found.add(id(objects)) found.add(id(found)) found.add(id(gc_objects)) for obj in objects: found.add(id(obj)) # repeatedly call get_referents until we can't find any more objects while len(nextObjList) > 0: curObjList = nextObjList nextObjList = [] for obj in curObjList: refs = gc.get_referents(obj) for ref in refs: if id(ref) not in found: found.add(id(ref)) objects.append(ref) nextObjList.append(ref) return objects