Source code for direct.showbase.PythonUtil

"""Contains miscellaneous utility functions and classes."""

__all__ = [

    'indent', 'doc', 'adjust', 'difference', 'intersection', 'union',
    'sameElements', 'makeList', 'makeTuple', 'list2dict', 'invertDict',
    'invertDictLossless', 'uniqueElements', 'disjoint', 'contains', 'replace',
    'reduceAngle', 'fitSrcAngle2Dest', 'fitDestAngle2Src', 'closestDestAngle2',
    'closestDestAngle', 'getSetterName', 'getSetter', 'Functor', 'Stack',
    'Queue', 'bound', 'clamp', 'lerp', 'average', 'addListsByValue',
    'boolEqual', 'lineupPos', 'formatElapsedSeconds', 'solveQuadratic',
    'findPythonModule', 'mostDerivedLast', 'clampScalar', 'weightedChoice',
    'randFloat', 'normalDistrib', 'weightedRand', 'randUint31', 'randInt32',
    'SerialNumGen', 'serialNum', 'uniqueName', 'Singleton',
    'SingletonError', 'printListEnum', 'safeRepr', 'fastRepr',
    'isDefaultValue', 'ScratchPad', 'Sync', 'itype', 'getNumberedTypedString',
    'getNumberedTypedSortedString', 'printNumberedTyped', 'DelayedCall',
    'DelayedFunctor', 'FrameDelayedCall', 'SubframeCall', 'getBase',
    'GoldenRatio', 'GoldenRectangle', 'rad90', 'rad180', 'rad270', 'rad360',
    'nullGen', 'loopGen', 'makeFlywheelGen', 'flywheel', 'listToIndex2item',
    'listToItem2index', 'formatTimeCompact', 'deeptype', 'StdoutCapture',
    'StdoutPassthrough', 'Averager', 'getRepository', 'formatTimeExact',
    'startSuperLog', 'endSuperLog', 'typeName', 'safeTypeName',
    'histogramDict', 'unescapeHtmlString',

if __debug__:
    __all__ += ['StackTrace', 'traceFunctionCall', 'traceParentCall',
                'printThisCall', 'stackEntryInfo', 'lineInfo', 'callerInfo',
                'lineTag', 'profileFunc', 'profiled', 'startProfile',
                'printProfile', 'getProfileResultString', 'printStack',

import types
import math
import os
import sys
import random
import time
import builtins
import importlib
import functools

__report_indent = 3

from panda3d.core import ConfigVariableBool, ClockObject

## with one integer positional arg, this uses about 4/5 of the memory of the Functor class below
#def Functor(function, *args, **kArgs):
#    argsCopy = args[:]
#    def functor(*cArgs, **ckArgs):
#        kArgs.update(ckArgs)
#        return function(*(argsCopy + cArgs), **kArgs)
#    return functor

[docs]class Functor:
[docs] def __init__(self, function, *args, **kargs): assert callable(function), "function should be a callable obj" self._function = function self._args = args self._kargs = kargs if hasattr(self._function, '__name__'): self.__name__ = self._function.__name__ else: self.__name__ = str(itype(self._function)) if hasattr(self._function, '__doc__'): self.__doc__ = self._function.__doc__ else: self.__doc__ = self.__name__
[docs] def destroy(self): del self._function del self._args del self._kargs del self.__name__ del self.__doc__
def _do__call__(self, *args, **kargs): _kargs = self._kargs.copy() _kargs.update(kargs) return self._function(*(self._args + args), **_kargs) __call__ = _do__call__ def __repr__(self): s = 'Functor(%s' % self._function.__name__ for arg in self._args: try: argStr = repr(arg) except: argStr = 'bad repr: %s' % arg.__class__ s += ', %s' % argStr for karg, value in list(self._kargs.items()): s += ', %s=%s' % (karg, repr(value)) s += ')' return s
[docs]class Stack:
[docs] def __init__(self): self.__list = []
[docs] def push(self, item): self.__list.append(item)
[docs] def top(self): # return the item on the top of the stack without popping it off return self.__list[-1]
[docs] def pop(self): return self.__list.pop()
[docs] def clear(self): self.__list = []
[docs] def isEmpty(self): return len(self.__list) == 0
def __len__(self): return len(self.__list)
[docs]class Queue: # FIFO queue # interface is intentionally identical to Stack (LIFO)
[docs] def __init__(self): self.__list = []
[docs] def push(self, item): self.__list.append(item)
[docs] def top(self): # return the next item at the front of the queue without popping it off return self.__list[0]
[docs] def front(self): return self.__list[0]
[docs] def back(self): return self.__list[-1]
[docs] def pop(self): return self.__list.pop(0)
[docs] def clear(self): self.__list = []
[docs] def isEmpty(self): return len(self.__list) == 0
def __len__(self): return len(self.__list)
[docs]def indent(stream, numIndents, str): """ Write str to stream with numIndents in front of it """ # To match emacs, instead of a tab character we will use 4 spaces stream.write(' ' * numIndents + str)
if __debug__: import traceback import marshal
[docs] class StackTrace:
[docs] def __init__(self, label="", start=0, limit=None): """ label is a string (or anything that be be a string) that is printed as part of the trace back. This is just to make it easier to tell what the stack trace is referring to. start is an integer number of stack frames back from the most recent. (This is automatically bumped up by one to skip the __init__ call to the StackTrace). limit is an integer number of stack frames to record (or None for unlimited). """ self.label = label if limit is not None: self.trace = traceback.extract_stack(sys._getframe(1+start), limit=limit) else: self.trace = traceback.extract_stack(sys._getframe(1+start))
[docs] def compact(self): r = '' comma = ',' for filename, lineNum, funcName, text in self.trace: r += '%s.%s:%s%s' % (filename[:filename.rfind('.py')][filename.rfind('\\')+1:], funcName, lineNum, comma) if len(r) > 0: r = r[:-len(comma)] return r
[docs] def reverseCompact(self): r = '' comma = ',' for filename, lineNum, funcName, text in self.trace: r = '%s.%s:%s%s%s' % (filename[:filename.rfind('.py')][filename.rfind('\\')+1:], funcName, lineNum, comma, r) if len(r) > 0: r = r[:-len(comma)] return r
def __str__(self): r = "Debug stack trace of %s (back %s frames):\n"%( self.label, len(self.trace),) for i in traceback.format_list(self.trace): r+=i r+="***** NOTE: This is not a crash. This is a debug stack trace. *****" return r
[docs] def printStack(): print(StackTrace(start=1).compact()) return True
[docs] def printReverseStack(): print(StackTrace(start=1).reverseCompact()) return True
[docs] def printVerboseStack(): print(StackTrace(start=1)) return True
[docs] def traceFunctionCall(frame): """ return a string that shows the call frame with calling arguments. e.g. foo(x=234, y=135) """ f = frame co = f.f_code dict = f.f_locals n = co.co_argcount if co.co_flags & 4: n = n + 1 if co.co_flags & 8: n = n + 1 r = '' if 'self' in dict: r = '%s.' % (dict['self'].__class__.__name__,) r += "%s(" % (f.f_code.co_name,) comma=0 # formatting, whether we should type a comma. for i in range(n): name = co.co_varnames[i] if name == 'self': continue if comma: r+=', ' else: # ok, we skipped the first one, the rest get commas: comma=1 r+=name r+='=' if name in dict: v=safeRepr(dict[name]) if len(v)>2000: # r+="<too big for debug>" r += (v[:2000] + "...") else: r+=v else: r+="*** undefined ***" return r+')'
[docs] def traceParentCall(): return traceFunctionCall(sys._getframe(2))
[docs] def printThisCall(): print(traceFunctionCall(sys._getframe(1))) return 1 # to allow "assert printThisCall()"
# Magic numbers: These are the bit masks in func_code.co_flags that # reveal whether or not the function has a *arg or **kw argument. _POS_LIST = 4 _KEY_DICT = 8
[docs]def doc(obj): if (isinstance(obj, types.MethodType)) or \ (isinstance(obj, types.FunctionType)): print(obj.__doc__)
[docs]def adjust(command = None, dim = 1, parent = None, **kw): """ adjust(command = None, parent = None, **kw) Popup and entry scale to adjust a parameter Accepts any Slider keyword argument. Typical arguments include: command: The one argument command to execute min: The min value of the slider max: The max value of the slider resolution: The resolution of the slider text: The label on the slider These values can be accessed and/or changed after the fact >>> vg = adjust() >>> vg['min'] 0.0 >>> vg['min'] = 10.0 >>> vg['min'] 10.0 """ # Make sure we enable Tk # Don't use a regular import, to prevent ModuleFinder from picking # it up as a dependency when building a .p3d package. Valuator = importlib.import_module('direct.tkwidgets.Valuator') # Set command if specified if command: kw['command'] = lambda x: command(*x) if parent is None: kw['title'] = command.__name__ kw['dim'] = dim # Create toplevel if needed if not parent: vg = Valuator.ValuatorGroupPanel(parent, **kw) else: vg = Valuator.ValuatorGroup(parent, **kw) vg.pack(expand = 1, fill = 'x') return vg
[docs]def difference(a, b): """ difference(list, list): """ if not a: return b if not b: return a d = [] for i in a: if (i not in b) and (i not in d): d.append(i) for i in b: if (i not in a) and (i not in d): d.append(i) return d
[docs]def intersection(a, b): """ intersection(list, list): """ if not a or not b: return [] d = [] for i in a: if (i in b) and (i not in d): d.append(i) for i in b: if (i in a) and (i not in d): d.append(i) return d
[docs]def union(a, b): """ union(list, list): """ # Copy a c = a[:] for i in b: if i not in c: c.append(i) return c
[docs]def sameElements(a, b): if len(a) != len(b): return 0 for elem in a: if elem not in b: return 0 for elem in b: if elem not in a: return 0 return 1
[docs]def makeList(x): """returns x, converted to a list""" if isinstance(x, list): return x elif isinstance(x, tuple): return list(x) else: return [x,]
[docs]def makeTuple(x): """returns x, converted to a tuple""" if isinstance(x, list): return tuple(x) elif isinstance(x, tuple): return x else: return (x,)
[docs]def list2dict(L, value=None): """creates dict using elements of list, all assigned to same value""" return dict([(k, value) for k in L])
[docs]def listToIndex2item(L): """converts list to dict of list index->list item""" d = {} for i, item in enumerate(L): d[i] = item return d
assert listToIndex2item(['a','b']) == {0: 'a', 1: 'b',}
[docs]def listToItem2index(L): """converts list to dict of list item->list index This is lossy if there are duplicate list items""" d = {} for i, item in enumerate(L): d[item] = i return d
assert listToItem2index(['a','b']) == {'a': 0, 'b': 1,}
[docs]def invertDict(D, lossy=False): """creates a dictionary by 'inverting' D; keys are placed in the new dictionary under their corresponding value in the old dictionary. It is an error if D contains any duplicate values. >>> old = {'key1':1, 'key2':2} >>> invertDict(old) {1: 'key1', 2: 'key2'} """ n = {} for key, value in D.items(): if not lossy and value in n: raise Exception('duplicate key in invertDict: %s' % value) n[value] = key return n
[docs]def invertDictLossless(D): """similar to invertDict, but values of new dict are lists of keys from old dict. No information is lost. >>> old = {'key1':1, 'key2':2, 'keyA':2} >>> invertDictLossless(old) {1: ['key1'], 2: ['key2', 'keyA']} """ n = {} for key, value in D.items(): n.setdefault(value, []) n[value].append(key) return n
[docs]def uniqueElements(L): """are all elements of list unique?""" return len(L) == len(list2dict(L))
[docs]def disjoint(L1, L2): """returns non-zero if L1 and L2 have no common elements""" used = dict([(k, None) for k in L1]) for k in L2: if k in used: return 0 return 1
[docs]def contains(whole, sub): """ Return 1 if whole contains sub, 0 otherwise """ if whole == sub: return 1 for elem in sub: # The first item you find not in whole, return 0 if elem not in whole: return 0 # If you got here, whole must contain sub return 1
[docs]def replace(list, old, new, all=0): """ replace 'old' with 'new' in 'list' if all == 0, replace first occurrence otherwise replace all occurrences returns the number of items replaced """ if old not in list: return 0 if not all: i = list.index(old) list[i] = new return 1 else: numReplaced = 0 for i in range(len(list)): if list[i] == old: numReplaced += 1 list[i] = new return numReplaced
rad90 = math.pi / 2. rad180 = math.pi rad270 = 1.5 * math.pi rad360 = 2. * math.pi
[docs]def reduceAngle(deg): """ Reduces an angle (in degrees) to a value in [-180..180) """ return ((deg + 180.) % 360.) - 180.
[docs]def fitSrcAngle2Dest(src, dest): """ given a src and destination angle, returns an equivalent src angle that is within [-180..180) of dest examples: fitSrcAngle2Dest(30, 60) == 30 fitSrcAngle2Dest(60, 30) == 60 fitSrcAngle2Dest(0, 180) == 0 fitSrcAngle2Dest(-1, 180) == 359 fitSrcAngle2Dest(-180, 180) == 180 """ return dest + reduceAngle(src - dest)
[docs]def fitDestAngle2Src(src, dest): """ given a src and destination angle, returns an equivalent dest angle that is within [-180..180) of src examples: fitDestAngle2Src(30, 60) == 60 fitDestAngle2Src(60, 30) == 30 fitDestAngle2Src(0, 180) == -180 fitDestAngle2Src(1, 180) == 180 """ return src + (reduceAngle(dest - src))
[docs]def closestDestAngle2(src, dest): # The function above didn't seem to do what I wanted. So I hacked # this one together. I can't really say I understand it. It's more # from impirical observation... GRW diff = src - dest if diff > 180: # if the difference is greater that 180 it's shorter to go the other way return dest - 360 elif diff < -180: # or perhaps the OTHER other way... return dest + 360 else: # otherwise just go to the original destination return dest
[docs]def closestDestAngle(src, dest): # The function above didn't seem to do what I wanted. So I hacked # this one together. I can't really say I understand it. It's more # from impirical observation... GRW diff = src - dest if diff > 180: # if the difference is greater that 180 it's shorter to go the other way return src - (diff - 360) elif diff < -180: # or perhaps the OTHER other way... return src - (360 + diff) else: # otherwise just go to the original destination return dest
[docs]class StdoutCapture: # redirects stdout to a string
[docs] def __init__(self): self._oldStdout = sys.stdout sys.stdout = self self._string = ''
[docs] def destroy(self): sys.stdout = self._oldStdout del self._oldStdout
[docs] def getString(self): return self._string
# internal
[docs] def write(self, string): self._string = ''.join([self._string, string])
[docs]class StdoutPassthrough(StdoutCapture): # like StdoutCapture but also allows output to go through to the OS as normal # internal
[docs] def write(self, string): self._string = ''.join([self._string, string]) self._oldStdout.write(string)
# constant profile defaults if __debug__: from io import StringIO PyUtilProfileDefaultFilename = 'profiledata' PyUtilProfileDefaultLines = 80 PyUtilProfileDefaultSorts = ['cumulative', 'time', 'calls'] _ProfileResultStr = ''
[docs] def getProfileResultString(): # if you called profile with 'log' not set to True, # you can call this function to get the results as # a string global _ProfileResultStr return _ProfileResultStr
[docs] def profileFunc(callback, name, terse, log=True): global _ProfileResultStr if 'globalProfileFunc' in builtins.__dict__: # rats. Python profiler is not re-entrant... base.notify.warning( 'PythonUtil.profileStart(%s): aborted, already profiling %s' #'\nStack Trace:\n%s' % (name, builtins.globalProfileFunc, #StackTrace() )) return builtins.globalProfileFunc = callback builtins.globalProfileResult = [None] prefix = '***** START PROFILE: %s *****' % name if log: print(prefix) startProfile(cmd='globalProfileResult[0]=globalProfileFunc()', callInfo=(not terse), silent=not log) suffix = '***** END PROFILE: %s *****' % name if log: print(suffix) else: _ProfileResultStr = '%s\n%s\n%s' % (prefix, _ProfileResultStr, suffix) result = globalProfileResult[0] del builtins.__dict__['globalProfileFunc'] del builtins.__dict__['globalProfileResult'] return result
[docs] def profiled(category=None, terse=False): """ decorator for profiling functions turn categories on and off via "want-profile-categoryName 1" e.g.:: @profiled('particles') def loadParticles(): ... :: want-profile-particles 1 """ assert type(category) in (str, type(None)), "must provide a category name for @profiled" # allow profiling in published versions #try: # null = not __dev__ #except: # null = not __debug__ #if null: # # if we're not in __dev__, just return the function itself. This # # results in zero runtime overhead, since decorators are evaluated # # at module-load. # def nullDecorator(f): # return f # return nullDecorator def profileDecorator(f): def _profiled(*args, **kArgs): name = '(%s) %s from %s' % (category, f.__name__, f.__module__) if category is None or ConfigVariableBool('want-profile-%s' % category, False).value: return profileFunc(Functor(f, *args, **kArgs), name, terse) else: return f(*args, **kArgs) _profiled.__doc__ = f.__doc__ return _profiled return profileDecorator
# intercept profile-related file operations to avoid disk access movedOpenFuncs = [] movedDumpFuncs = [] movedLoadFuncs = [] profileFilenames = set() profileFilenameList = Stack() profileFilename2file = {} profileFilename2marshalData = {} def _profileOpen(filename, *args, **kArgs): # this is a replacement for the file open() builtin function # for use during profiling, to intercept the file open # operation used by the Python profiler and profile stats # systems if filename in profileFilenames: # if this is a file related to profiling, create an # in-RAM file object if filename not in profileFilename2file: file = StringIO() file._profFilename = filename profileFilename2file[filename] = file else: file = profileFilename2file[filename] else: file = movedOpenFuncs[-1](filename, *args, **kArgs) return file def _profileMarshalDump(data, file): # marshal.dump doesn't work with StringIO objects # simulate it if isinstance(file, StringIO) and hasattr(file, '_profFilename'): if file._profFilename in profileFilenames: profileFilename2marshalData[file._profFilename] = data return None return movedDumpFuncs[-1](data, file) def _profileMarshalLoad(file): # marshal.load doesn't work with StringIO objects # simulate it if isinstance(file, StringIO) and hasattr(file, '_profFilename'): if file._profFilename in profileFilenames: return profileFilename2marshalData[file._profFilename] return movedLoadFuncs[-1](file) def _installProfileCustomFuncs(filename): assert filename not in profileFilenames profileFilenames.add(filename) profileFilenameList.push(filename) movedOpenFuncs.append( = _profileOpen movedDumpFuncs.append(marshal.dump) marshal.dump = _profileMarshalDump movedLoadFuncs.append(marshal.load) marshal.load = _profileMarshalLoad def _getProfileResultFileInfo(filename): return (profileFilename2file.get(filename, None), profileFilename2marshalData.get(filename, None)) def _setProfileResultsFileInfo(filename, info): f, m = info if f: profileFilename2file[filename] = f if m: profileFilename2marshalData[filename] = m def _clearProfileResultFileInfo(filename): profileFilename2file.pop(filename, None) profileFilename2marshalData.pop(filename, None) def _removeProfileCustomFuncs(filename): assert == filename marshal.load = movedLoadFuncs.pop() marshal.dump = movedDumpFuncs.pop() = movedOpenFuncs.pop() profileFilenames.remove(filename) profileFilenameList.pop() profileFilename2file.pop(filename, None) # don't let marshalled data pile up profileFilename2marshalData.pop(filename, None) # call this from the prompt, and break back out to the prompt # to stop profiling # # OR to do inline profiling, you must make a globally-visible # function to be profiled, i.e. to profile 'self.load()', do # something like this: # # def func(self=self): # self.load() # import builtins # builtins.func = func # PythonUtil.startProfile(cmd='func()', filename='profileData') # del builtins.func # def _profileWithoutGarbageLeak(cmd, filename): # The profile module isn't necessarily installed on every Python # installation, so we import it here, instead of in the module # scope. import profile # this is necessary because the profile module creates a memory leak Profile = profile.Profile statement = cmd sort = -1 #### COPIED FROM #### prof = Profile() try: prof = except SystemExit: pass if filename is not None: prof.dump_stats(filename) else: #return prof.print_stats(sort) #DCR prof.print_stats(sort) #DCR ################################# # eliminate the garbage leak del prof.dispatcher
[docs] def startProfile(filename=PyUtilProfileDefaultFilename, lines=PyUtilProfileDefaultLines, sorts=PyUtilProfileDefaultSorts, silent=0, callInfo=1, useDisk=False, cmd='run()'): # uniquify the filename to allow multiple processes to profile simultaneously filename = '%s.%s%s' % (filename, randUint31(), randUint31()) if not useDisk: # use a RAM file _installProfileCustomFuncs(filename) _profileWithoutGarbageLeak(cmd, filename) if silent: extractProfile(filename, lines, sorts, callInfo) else: printProfile(filename, lines, sorts, callInfo) if not useDisk: # discard the RAM file _removeProfileCustomFuncs(filename) else: os.remove(filename)
# call these to see the results again, as a string or in the log
[docs] def printProfile(filename=PyUtilProfileDefaultFilename, lines=PyUtilProfileDefaultLines, sorts=PyUtilProfileDefaultSorts, callInfo=1): import pstats s = pstats.Stats(filename) s.strip_dirs() for sort in sorts: s.sort_stats(sort) s.print_stats(lines) if callInfo: s.print_callees(lines) s.print_callers(lines)
# same args as printProfile
[docs] def extractProfile(*args, **kArgs): global _ProfileResultStr # capture print output sc = StdoutCapture() # print the profile output, redirected to the result string printProfile(*args, **kArgs) # make a copy of the print output _ProfileResultStr = sc.getString() # restore stdout to what it was before sc.destroy()
[docs]def getSetterName(valueName, prefix='set'): # getSetterName('color') -> 'setColor' # getSetterName('color', 'get') -> 'getColor' return '%s%s%s' % (prefix, valueName[0].upper(), valueName[1:])
[docs]def getSetter(targetObj, valueName, prefix='set'): # getSetter(smiley, 'pos') -> smiley.setPos return getattr(targetObj, getSetterName(valueName, prefix))
[docs]def mostDerivedLast(classList): """pass in list of classes. sorts list in-place, with derived classes appearing after their bases""" class ClassSortKey(object): __slots__ = 'classobj', def __init__(self, classobj): self.classobj = classobj def __lt__(self, other): return issubclass(other.classobj, self.classobj) classList.sort(key=ClassSortKey)
[docs]def bound(value, bound1, bound2): """ returns value if value is between bound1 and bound2 otherwise returns bound that is closer to value """ if bound1 > bound2: return min(max(value, bound2), bound1) else: return min(max(value, bound1), bound2)
clamp = bound
[docs]def lerp(v0, v1, t): """ returns a value lerped between v0 and v1, according to t t == 0 maps to v0, t == 1 maps to v1 """ return v0 + ((v1 - v0) * t)
[docs]def getShortestRotation(start, end): """ Given two heading values, return a tuple describing the shortest interval from 'start' to 'end'. This tuple can be used to lerp a camera between two rotations while avoiding the 'spin' problem. """ start, end = start % 360, end % 360 if abs(end - start) > 180: if end < start: end += 360 else: start += 360 return (start, end)
[docs]def average(*args): """ returns simple average of list of values """ val = 0. for arg in args: val += arg return val / len(args)
[docs]class Averager:
[docs] def __init__(self, name): self._name = name self.reset()
[docs] def reset(self): self._total = 0. self._count = 0
[docs] def addValue(self, value): self._total += value self._count += 1
[docs] def getAverage(self): return self._total / self._count
[docs] def getCount(self): return self._count
[docs]def addListsByValue(a, b): """ returns a new array containing the sums of the two array arguments (c[0] = a[0 + b[0], etc.) """ c = [] for x, y in zip(a, b): c.append(x + y) return c
[docs]def boolEqual(a, b): """ returns true if a and b are both true or both false. returns false otherwise (a.k.a. xnor -- eXclusive Not OR). """ return (a and b) or not (a or b)
[docs]def lineupPos(i, num, spacing): """ use to line up a series of 'num' objects, in one dimension, centered around zero 'i' is the index of the object in the lineup 'spacing' is the amount of space between objects in the lineup """ assert num >= 1 assert i >= 0 and i < num pos = float(i) * spacing return pos - ((float(spacing) * (num-1))/2.)
[docs]def formatElapsedSeconds(seconds): """ Returns a string of the form "mm:ss" or "hh:mm:ss" or "n days", representing the indicated elapsed time in seconds. """ sign = '' if seconds < 0: seconds = -seconds sign = '-' # We use math.floor() instead of casting to an int, so we avoid # problems with numbers that are too large to represent as # type int. seconds = math.floor(seconds) hours = math.floor(seconds / (60 * 60)) if hours > 36: days = math.floor((hours + 12) / 24) return "%s%d days" % (sign, days) seconds -= hours * (60 * 60) minutes = (int)(seconds / 60) seconds -= minutes * 60 if hours != 0: return "%s%d:%02d:%02d" % (sign, hours, minutes, seconds) else: return "%s%d:%02d" % (sign, minutes, seconds)
[docs]def solveQuadratic(a, b, c): # quadratic equation: ax^2 + bx + c = 0 # quadratic formula: x = [-b +/- sqrt(b^2 - 4ac)] / 2a # returns None, root, or [root1, root2] # a cannot be zero. if a == 0.: return None # calculate the determinant (b^2 - 4ac) D = (b * b) - (4. * a * c) if D < 0: # there are no solutions (sqrt(negative number) is undefined) return None elif D == 0: # only one root return (-b) / (2. * a) else: # OK, there are two roots sqrtD = math.sqrt(D) twoA = 2. * a root1 = ((-b) - sqrtD) / twoA root2 = ((-b) + sqrtD) / twoA return [root1, root2]
if __debug__:
[docs] def stackEntryInfo(depth=0, baseFileName=1): """ returns the sourcefilename, line number, and function name of an entry in the stack. 'depth' is how far back to go in the stack; 0 is the caller of this function, 1 is the function that called the caller of this function, etc. by default, strips off the path of the filename; override with baseFileName returns (fileName, lineNum, funcName) --> (string, int, string) returns (None, None, None) on error """ import inspect try: stack = None frame = None try: stack = inspect.stack() # add one to skip the frame associated with this function frame = stack[depth+1] filename = frame[1] if baseFileName: filename = os.path.basename(filename) lineNum = frame[2] funcName = frame[3] result = (filename, lineNum, funcName) finally: del stack del frame except: result = (None, None, None) return result
[docs] def lineInfo(baseFileName=1): """ returns the sourcefilename, line number, and function name of the code that called this function (answers the question: 'hey lineInfo, where am I in the codebase?') see stackEntryInfo, above, for info on 'baseFileName' and return types """ return stackEntryInfo(1, baseFileName)
[docs] def callerInfo(baseFileName=1, howFarBack=0): """ returns the sourcefilename, line number, and function name of the caller of the function that called this function (answers the question: 'hey callerInfo, who called me?') see stackEntryInfo, above, for info on 'baseFileName' and return types """ return stackEntryInfo(2+howFarBack, baseFileName)
[docs] def lineTag(baseFileName=1, verbose=0, separator=':'): """ returns a string containing the sourcefilename and line number of the code that called this function (equivalent to lineInfo, above, with different return type) see stackEntryInfo, above, for info on 'baseFileName' if 'verbose' is false, returns a compact string of the form 'fileName:lineNum:funcName' if 'verbose' is true, returns a longer string that matches the format of Python stack trace dumps returns empty string on error """ fileName, lineNum, funcName = callerInfo(baseFileName) if fileName is None: return '' if verbose: return 'File "%s", line %s, in %s' % (fileName, lineNum, funcName) else: return '%s%s%s%s%s' % (fileName, separator, lineNum, separator, funcName)
[docs]def findPythonModule(module): # Look along the python load path for the indicated filename. # Returns the located pathname, or None if the filename is not # found. filename = module + '.py' for dir in sys.path: pathname = os.path.join(dir, filename) if os.path.exists(pathname): return pathname return None
[docs]def clampScalar(value, a, b): # calling this ought to be faster than calling both min and max if a < b: if value < a: return a elif value > b: return b else: return value else: if value < b: return b elif value > a: return a else: return value
[docs]def weightedChoice(choiceList, rng=random.random, sum=None): """given a list of (weight, item) pairs, chooses an item based on the weights. rng must return 0..1. if you happen to have the sum of the weights, pass it in 'sum'.""" # Throw an IndexError if we got an empty list. if not choiceList: raise IndexError('Cannot choose from an empty sequence') # TODO: add support for dicts if sum is None: sum = 0. for weight, item in choiceList: sum += weight rand = rng() accum = rand * sum item = None for weight, item in choiceList: accum -= weight if accum <= 0.: return item # rand is ~1., and floating-point error prevented accum from hitting 0. # Or you passed in a 'sum' that was was too large. # Return the last item. return item
[docs]def randFloat(a, b=0., rng=random.random): """returns a random float in [a, b] call with single argument to generate random float between arg and zero """ return lerp(a, b, rng())
[docs]def normalDistrib(a, b, gauss=random.gauss): """ NOTE: assumes a < b Returns random number between a and b, using gaussian distribution, with mean=avg(a, b), and a standard deviation that fits ~99.7% of the curve between a and b. For ease of use, outlying results are re-computed until result is in [a, b] This should fit the remaining .3% of the curve that lies outside [a, b] uniformly onto the curve inside [a, b] ------------------------------------------------------------------------ The 68-95-99.7% Rule ==================== All normal density curves satisfy the following property which is often referred to as the Empirical Rule: 68% of the observations fall within 1 standard deviation of the mean. 95% of the observations fall within 2 standard deviations of the mean. 99.7% of the observations fall within 3 standard deviations of the mean. Thus, for a normal distribution, almost all values lie within 3 standard deviations of the mean. ------------------------------------------------------------------------ In calculating our standard deviation, we divide (b-a) by 6, since the 99.7% figure includes 3 standard deviations _on_either_side_ of the mean. """ while True: r = gauss((a+b)*.5, (b-a)/6.) if (r >= a) and (r <= b): return r
[docs]def weightedRand(valDict, rng=random.random): """ pass in a dictionary with a selection -> weight mapping. E.g.:: {"Choice 1": 10, "Choice 2": 30, "bear": 100} - Weights need not add up to any particular value. - The actual selection will be returned. """ selections = list(valDict.keys()) weights = list(valDict.values()) totalWeight = 0 for weight in weights: totalWeight += weight # get a random value between 0 and the total of the weights randomWeight = rng() * totalWeight # find the index that corresponds with this weight for i in range(len(weights)): totalWeight -= weights[i] if totalWeight <= randomWeight: return selections[i] assert True, "Should never get here" return selections[-1]
[docs]def randUint31(rng=random.random): """returns a random integer in [0..2^31). rng must return float in [0..1]""" return int(rng() * 0x7FFFFFFF)
[docs]def randInt32(rng=random.random): """returns a random integer in [-2147483648..2147483647]. rng must return float in [0..1] """ i = int(rng() * 0x7FFFFFFF) if rng() < .5: i *= -1 return i
[docs]class SerialNumGen: """generates serial numbers"""
[docs] def __init__(self, start=None): if start is None: start = 0 self.__counter = start-1
[docs] def next(self): self.__counter += 1 return self.__counter
__next__ = next
[docs]class SerialMaskedGen(SerialNumGen):
[docs] def __init__(self, mask, start=None): self._mask = mask SerialNumGen.__init__(self, start)
[docs] def next(self): v = return v & self._mask
__next__ = next
_serialGen = SerialNumGen()
[docs]def serialNum(): global _serialGen return
[docs]def uniqueName(name): global _serialGen return '%s-%s' % (name,
############################################################ # class: Singleton # Purpose: This provides a base metaclass for all classes # that require one and only one instance. # # Example: class mySingleton: # __metaclass__ = PythonUtil.Singleton # def __init__(self, ...): # ... # # Note: This class is based on Python's New-Style Class # design. An error will occur if a defined class # attemps to inherit from a Classic-Style Class only, # ie: class myClassX: # def __init__(self, ...): # ... # # class myNewClassX(myClassX): # __metaclass__ = PythonUtil.Singleton # def __init__(self, ...): # myClassX.__init__(self, ...) # ... # # This causes problems because myNewClassX is a # New-Style class that inherits from only a # Classic-Style base class. There are two ways # simple ways to resolve this issue. # # First, if possible, make myClassX a # New-Style class by inheriting from object # object. IE: class myClassX(object): # # If for some reason that is not an option, make # myNewClassX inherit from object and myClassX. # IE: class myNewClassX(object, myClassX): ############################################################
[docs]class Singleton(type):
[docs] def __init__(cls, name, bases, dic): super(Singleton, cls).__init__(name, bases, dic) cls.instance=None
def __call__(cls, *args, **kw): if cls.instance is None: cls.instance=super(Singleton, cls).__call__(*args, **kw) return cls.instance
[docs]class SingletonError(ValueError): """ Used to indicate an inappropriate value for a Singleton."""
[docs]def printListEnumGen(l): # log each individual item with a number in front of it digits = 0 n = len(l) while n > 0: digits += 1 n //= 10 format = '%0' + '%s' % digits + 'i:%s' for i in range(len(l)): print(format % (i, l[i])) yield None
[docs]def printListEnum(l): for result in printListEnumGen(l): pass
# base class for all Panda C++ objects # libdtoolconfig doesn't seem to have this, grab it off of TypedObject dtoolSuperBase = None def _getDtoolSuperBase(): global dtoolSuperBase from panda3d.core import TypedObject dtoolSuperBase = TypedObject.__bases__[0] assert dtoolSuperBase.__name__ == 'DTOOL_SUPER_BASE' safeReprNotify = None def _getSafeReprNotify(): global safeReprNotify from direct.directnotify.DirectNotifyGlobal import directNotify safeReprNotify = directNotify.newCategory("safeRepr") return safeReprNotify
[docs]def safeRepr(obj): global dtoolSuperBase if dtoolSuperBase is None: _getDtoolSuperBase() global safeReprNotify if safeReprNotify is None: _getSafeReprNotify() if isinstance(obj, dtoolSuperBase): # repr of C++ object could crash, particularly if the object has been deleted # log that we're calling repr'calling repr on instance of %s.%s' % (obj.__class__.__module__, obj.__class__.__name__)) sys.stdout.flush() try: return repr(obj) except: return '<** FAILED REPR OF %s instance at %s **>' % (obj.__class__.__name__, hex(id(obj)))
[docs]def safeReprTypeOnFail(obj): global dtoolSuperBase if dtoolSuperBase is None: _getDtoolSuperBase() global safeReprNotify if safeReprNotify is None: _getSafeReprNotify() if isinstance(obj, dtoolSuperBase): return type(obj) try: return repr(obj) except: return '<** FAILED REPR OF %s instance at %s **>' % (obj.__class__.__name__, hex(id(obj)))
[docs]def fastRepr(obj, maxLen=200, strFactor=10, _visitedIds=None): """ caps the length of iterable types, so very large objects will print faster. also prevents infinite recursion """ try: if _visitedIds is None: _visitedIds = set() if id(obj) in _visitedIds: return '<ALREADY-VISITED %s>' % itype(obj) if type(obj) in (tuple, list): s = '' s += {tuple: '(', list: '[',}[type(obj)] if maxLen is not None and len(obj) > maxLen: o = obj[:maxLen] ellips = '...' else: o = obj ellips = '' _visitedIds.add(id(obj)) for item in o: s += fastRepr(item, maxLen, _visitedIds=_visitedIds) s += ', ' _visitedIds.remove(id(obj)) s += ellips s += {tuple: ')', list: ']',}[type(obj)] return s elif type(obj) is dict: s = '{' if maxLen is not None and len(obj) > maxLen: o = list(obj.keys())[:maxLen] ellips = '...' else: o = list(obj.keys()) ellips = '' _visitedIds.add(id(obj)) for key in o: value = obj[key] s += '%s: %s, ' % (fastRepr(key, maxLen, _visitedIds=_visitedIds), fastRepr(value, maxLen, _visitedIds=_visitedIds)) _visitedIds.remove(id(obj)) s += ellips s += '}' return s elif type(obj) is str: if maxLen is not None: maxLen *= strFactor if maxLen is not None and len(obj) > maxLen: return safeRepr(obj[:maxLen]) else: return safeRepr(obj) else: r = safeRepr(obj) maxLen *= strFactor if len(r) > maxLen: r = r[:maxLen] return r except: return '<** FAILED REPR OF %s **>' % obj.__class__.__name__
[docs]def convertTree(objTree, idList): newTree = {} for key in list(objTree.keys()): obj = (idList[key],) newTree[obj] = {} r_convertTree(objTree[key], newTree[obj], idList) return newTree
[docs]def r_convertTree(oldTree, newTree, idList): for key in list(oldTree.keys()): obj = idList.get(key) if not obj: continue obj = str(obj)[:100] newTree[obj] = {} r_convertTree(oldTree[key], newTree[obj], idList)
[docs]def pretty_print(tree): for name in tree.keys(): print(name) r_pretty_print(tree[name], 0)
[docs]def r_pretty_print(tree, num): num += 1 for name in tree.keys(): print(" " * num, name) r_pretty_print(tree[name], num)
[docs]def isDefaultValue(x): return x == type(x)()
[docs]def appendStr(obj, st): """adds a string onto the __str__ output of an instance""" def appendedStr(oldStr, st, self): return oldStr() + st oldStr = getattr(obj, '__str__', None) if oldStr is None: def stringer(s): return s oldStr = Functor(stringer, str(obj)) stringer = None obj.__str__ = types.MethodType(Functor(appendedStr, oldStr, st), obj) appendedStr = None return obj
[docs]class ScratchPad: """empty class to stick values onto"""
[docs] def __init__(self, **kArgs): for key, value in kArgs.items(): setattr(self, key, value) self._keys = set(kArgs.keys())
[docs] def add(self, **kArgs): for key, value in kArgs.items(): setattr(self, key, value) self._keys.update(list(kArgs.keys()))
[docs] def destroy(self): for key in self._keys: delattr(self, key)
# allow dict [] syntax def __getitem__(self, itemName): return getattr(self, itemName)
[docs] def get(self, itemName, default=None): return getattr(self, itemName, default)
# allow 'in' def __contains__(self, itemName): return itemName in self._keys
[docs]class Sync: _SeriesGen = SerialNumGen()
[docs] def __init__(self, name, other=None): self._name = name if other is None: self._series = self._value = 0 else: self._series = other._series self._value = other._value
[docs] def invalidate(self): self._value = None
[docs] def change(self): self._value += 1
[docs] def sync(self, other): if (self._series != other._series) or (self._value != other._value): self._series = other._series self._value = other._value return True else: return False
[docs] def isSynced(self, other): return ((self._series == other._series) and (self._value == other._value))
def __repr__(self): return '%s(%s)<family=%s,value=%s>' % (self.__class__.__name__, self._name, self._series, self._value)
[docs]def itype(obj): # version of type that gives more complete information about instance types global dtoolSuperBase t = type(obj) # C++ object instances appear to be types via type() # check if this is a C++ object if dtoolSuperBase is None: _getDtoolSuperBase() if isinstance(obj, dtoolSuperBase): return "<type 'instance' of %s>" % (obj.__class__) return t
[docs]def deeptype(obj, maxLen=100, _visitedIds=None): if _visitedIds is None: _visitedIds = set() if id(obj) in _visitedIds: return '<ALREADY-VISITED %s>' % itype(obj) t = type(obj) if t in (tuple, list): s = '' s += {tuple: '(', list: '[',}[type(obj)] if maxLen is not None and len(obj) > maxLen: o = obj[:maxLen] ellips = '...' else: o = obj ellips = '' _visitedIds.add(id(obj)) for item in o: s += deeptype(item, maxLen, _visitedIds=_visitedIds) s += ', ' _visitedIds.remove(id(obj)) s += ellips s += {tuple: ')', list: ']',}[type(obj)] return s elif type(obj) is dict: s = '{' if maxLen is not None and len(obj) > maxLen: o = list(obj.keys())[:maxLen] ellips = '...' else: o = list(obj.keys()) ellips = '' _visitedIds.add(id(obj)) for key in o: value = obj[key] s += '%s: %s, ' % (deeptype(key, maxLen, _visitedIds=_visitedIds), deeptype(value, maxLen, _visitedIds=_visitedIds)) _visitedIds.remove(id(obj)) s += ellips s += '}' return s else: return str(itype(obj))
[docs]def getNumberedTypedString(items, maxLen=5000, numPrefix=''): """get a string that has each item of the list on its own line, and each item is numbered on the left from zero""" digits = 0 n = len(items) while n > 0: digits += 1 n //= 10 format = numPrefix + '%0' + '%s' % digits + 'i:%s \t%s' first = True s = '' snip = '<SNIP>' for i in range(len(items)): if not first: s += '\n' first = False objStr = fastRepr(items[i]) if len(objStr) > maxLen: objStr = '%s%s' % (objStr[:(maxLen-len(snip))], snip) s += format % (i, itype(items[i]), objStr) return s
[docs]def getNumberedTypedSortedString(items, maxLen=5000, numPrefix=''): """get a string that has each item of the list on its own line, the items are stringwise-sorted, and each item is numbered on the left from zero""" digits = 0 n = len(items) while n > 0: digits += 1 n //= 10 format = numPrefix + '%0' + '%s' % digits + 'i:%s \t%s' snip = '<SNIP>' strs = [] for item in items: objStr = fastRepr(item) if len(objStr) > maxLen: objStr = '%s%s' % (objStr[:(maxLen-len(snip))], snip) strs.append(objStr) first = True s = '' strs.sort() for i in range(len(strs)): if not first: s += '\n' first = False objStr = strs[i] s += format % (i, itype(items[i]), strs[i]) return s
[docs]def printNumberedTyped(items, maxLen=5000): """print out each item of the list on its own line, with each item numbered on the left from zero""" digits = 0 n = len(items) while n > 0: digits += 1 n //= 10 format = '%0' + '%s' % digits + 'i:%s \t%s' for i in range(len(items)): objStr = fastRepr(items[i]) if len(objStr) > maxLen: snip = '<SNIP>' objStr = '%s%s' % (objStr[:(maxLen-len(snip))], snip) print(format % (i, itype(items[i]), objStr))
[docs]def printNumberedTypesGen(items, maxLen=5000): digits = 0 n = len(items) while n > 0: digits += 1 n //= 10 format = '%0' + '%s' % digits + 'i:%s' for i in range(len(items)): print(format % (i, itype(items[i]))) yield None
[docs]def printNumberedTypes(items, maxLen=5000): """print out the type of each item of the list on its own line, with each item numbered on the left from zero""" for result in printNumberedTypesGen(items, maxLen): yield result
[docs]class DelayedCall: """ calls a func after a specified delay """
[docs] def __init__(self, func, name=None, delay=None): if name is None: name = 'anonymous' if delay is None: delay = .01 self._func = func self._taskName = 'DelayedCallback-%s' % name self._delay = delay self._finished = False self._addDoLater()
[docs] def destroy(self): self._finished = True self._removeDoLater()
[docs] def finish(self): if not self._finished: self._doCallback() self.destroy()
def _addDoLater(self): taskMgr.doMethodLater(self._delay, self._doCallback, self._taskName) def _removeDoLater(self): taskMgr.remove(self._taskName) def _doCallback(self, task): self._finished = True func = self._func del self._func func()
[docs]class FrameDelayedCall: """ calls a func after N frames """
[docs] def __init__(self, name, callback, frames=None, cancelFunc=None): # checkFunc is optional; called every frame, if returns True, FrameDelay is cancelled # and callback is not called if frames is None: frames = 1 self._name = name self._frames = frames self._callback = callback self._cancelFunc = cancelFunc self._taskName = uniqueName('%s-%s' % (self.__class__.__name__, self._name)) self._finished = False self._startTask()
[docs] def destroy(self): self._finished = True self._stopTask()
[docs] def finish(self): if not self._finished: self._finished = True self._callback() self.destroy()
def _startTask(self): taskMgr.add(self._frameTask, self._taskName) self._counter = 0 def _stopTask(self): taskMgr.remove(self._taskName) def _frameTask(self, task): if self._cancelFunc and self._cancelFunc(): self.destroy() return task.done self._counter += 1 if self._counter >= self._frames: self.finish() return task.done return task.cont
[docs]class DelayedFunctor: """ Waits for this object to be called, then calls supplied functor after a delay. Effectively inserts a time delay between the caller and the functor. """
[docs] def __init__(self, functor, name=None, delay=None): self._functor = functor self._name = name # FunctionInterval requires __name__ self.__name__ = self._name self._delay = delay
def _callFunctor(self): cb = Functor(self._functor, *self._args, **self._kwArgs) del self._functor del self._name del self._delay del self._args del self._kwArgs del self._delayedCall del self.__name__ cb() def __call__(self, *args, **kwArgs): self._args = args self._kwArgs = kwArgs self._delayedCall = DelayedCall(self._callFunctor, self._name, self._delay)
[docs]class SubframeCall: """Calls a callback at a specific time during the frame using the task system"""
[docs] def __init__(self, functor, taskPriority, name=None): self._functor = functor self._name = name self._taskName = uniqueName('SubframeCall-%s' % self._name) taskMgr.add(self._doCallback, self._taskName, priority=taskPriority)
def _doCallback(self, task): functor = self._functor del self._functor functor() del self._name self._taskName = None return task.done
[docs] def cleanup(self): if self._taskName: taskMgr.remove(self._taskName) self._taskName = None
[docs]class PStatScope: collectors = {}
[docs] def __init__(self, level = None): self.levels = [] if level: self.levels.append(level)
[docs] def copy(self, push = None): c = PStatScope() c.levels = self.levels[:] if push: c.push(push) return c
def __repr__(self): return 'PStatScope - \'%s\'' % (self,) def __str__(self): return ':'.join(self.levels)
[docs] def push(self, level): self.levels.append(level.replace('_',''))
[docs] def pop(self): return self.levels.pop()
[docs] def start(self, push = None): if push: self.push(push) self.getCollector().start()
[docs] def stop(self, pop = False): self.getCollector().stop() if pop: self.pop()
[docs] def getCollector(self): label = str(self) if label not in self.collectors: from panda3d.core import PStatCollector self.collectors[label] = PStatCollector(label) # print ' ',self.collectors[label] return self.collectors[label]
[docs]def pstatcollect(scope, level = None): def decorator(f): return f try: if not (__dev__ or ConfigVariableBool('force-pstatcollect', False)) or \ not scope: return decorator def decorator(f): def wrap(*args, **kw): scope.start(push = (level or f.__name__)) val = f(*args, **kw) scope.stop(pop = True) return val return wrap except: pass return decorator
__report_indent = 0
[docs]def report(types = [], prefix = '', xform = None, notifyFunc = None, dConfigParam = []): """ This is a decorator generating function. Use is similar to a @decorator, except you must be sure to call it as a function. It actually returns the decorator which is then used to transform your decorated function. Confusing at first, I know. Decoration occurs at function definition time. If __dev__ is not defined, or resolves to False, this function has no effect and no wrapping/transform occurs. So in production, it's as if the report has been asserted out. Parameters: types: A subset list of ['timeStamp', 'frameCount', 'avLocation'] This allows you to specify certain useful bits of info: - *module*: Prints the module that this report statement can be found in. - *args*: Prints the arguments as they were passed to this function. - *timeStamp*: Adds the current frame time to the output. - *deltaStamp*: Adds the current AI synched frame time to the output - *frameCount*: Adds the current frame count to the output. Usually cleaner than the timeStamp output. - *avLocation*: Adds the localAvatar's network location to the output. Useful for interest debugging. - *interests*: Prints the current interest state after the report. - *stackTrace*: Prints a stack trace after the report. prefix: Optional string to prepend to output, just before the function. Allows for easy grepping and is useful when merging AI/Client reports into a single file. xform: Optional callback that accepts a single parameter: argument 0 to the decorated function. (assumed to be 'self') It should return a value to be inserted into the report output string. notifyFunc: A notify function such as info, debug, warning, etc. By default the report will be printed to stdout. This will allow you send the report to a designated 'notify' output. dConfigParam: A list of Config.prc string variables. By default the report will always print. If you specify this param, it will only print if one of the specified config strings resolve to True. """ def indent(str): global __report_indent return ' '*__report_indent+str def decorator(f): return f try: if not __dev__ and not ConfigVariableBool('force-reports', False): return decorator # determine whether we should use the decorator # based on the value of dConfigParam. dConfigParamList = [] doPrint = False if not dConfigParam: doPrint = True else: if not isinstance(dConfigParam, (list,tuple)): dConfigParams = (dConfigParam,) else: dConfigParams = dConfigParam dConfigParamList = [param for param in dConfigParams \ if ConfigVariableBool('want-%s-report' % (param,), False)] doPrint = bool(dConfigParamList) if not doPrint: return decorator # Determine any prefixes defined in our Config.prc. if prefix: prefixes = set([prefix]) else: prefixes = set() for param in dConfigParamList: prefix = config.GetString('prefix-%s-report' % (param,), '') if prefix: prefixes.add(prefix) except NameError as e: return decorator globalClockDelta = importlib.import_module("direct.distributed.ClockDelta").globalClockDelta def decorator(f): def wrap(*args,**kwargs): if args: rArgs = [args[0].__class__.__name__ + ', '] else: rArgs = [] if 'args' in types: rArgs += [repr(x)+', ' for x in args[1:]] + \ [ x + ' = ' + '%s, ' % repr(y) for x,y in kwargs.items()] if not rArgs: rArgs = '()' else: rArgs = '(' + functools.reduce(str.__add__,rArgs)[:-2] + ')' outStr = '%s%s' % (f.__name__, rArgs) # Insert prefix place holder, if needed if prefixes: outStr = '%%s %s' % (outStr,) globalClock = ClockObject.getGlobalClock() if 'module' in types: outStr = '%s {M:%s}' % (outStr, f.__module__.split('.')[-1]) if 'frameCount' in types: outStr = '%-8d : %s' % (globalClock.getFrameCount(), outStr) if 'timeStamp' in types: outStr = '%-8.3f : %s' % (globalClock.getFrameTime(), outStr) if 'deltaStamp' in types: outStr = '%-8.2f : %s' % (globalClock.getRealTime() - \, outStr) if 'avLocation' in types: outStr = '%s : %s' % (outStr, str(localAvatar.getLocation())) if xform: outStr = '%s : %s' % (outStr, xform(args[0])) if prefixes: # This will print the same report once for each prefix for prefix in prefixes: if notifyFunc: notifyFunc(outStr % (prefix,)) else: print(indent(outStr % (prefix,))) else: if notifyFunc: notifyFunc(outStr) else: print(indent(outStr)) if 'interests' in types: if 'stackTrace' in types: print(StackTrace()) global __report_indent rVal = None try: __report_indent += 1 rVal = f(*args,**kwargs) finally: __report_indent -= 1 if rVal is not None: print(indent(' -> '+repr(rVal))) return rVal wrap.__name__ = f.__name__ wrap.__dict__ = f.__dict__ wrap.__doc__ = f.__doc__ wrap.__module__ = f.__module__ return wrap return decorator
[docs]def getBase(): try: return base except: return simbase
[docs]def getRepository(): try: return except: return simbase.air
exceptionLoggedNotify = None if __debug__:
[docs] def exceptionLogged(append=True): """decorator that outputs the function name and all arguments if an exception passes back through the stack frame if append is true, string is appended to the __str__ output of the exception. if append is false, string is printed to the log directly. If the output will take up many lines, it's recommended to set append to False so that the exception stack is not hidden by the output of this decorator. """ try: null = not __dev__ except: null = not __debug__ if null: # if we're not in __dev__, just return the function itself. This # results in zero runtime overhead, since decorators are evaluated # at module-load. def nullDecorator(f): return f return nullDecorator def _decoratorFunc(f, append=append): global exceptionLoggedNotify if exceptionLoggedNotify is None: from direct.directnotify.DirectNotifyGlobal import directNotify exceptionLoggedNotify = directNotify.newCategory("ExceptionLogged") def _exceptionLogged(*args, **kArgs): try: return f(*args, **kArgs) except Exception as e: try: s = '%s(' % f.__name__ for arg in args: s += '%s, ' % arg for key, value in list(kArgs.items()): s += '%s=%s, ' % (key, value) if len(args) > 0 or len(kArgs) > 0: s = s[:-2] s += ')' if append: appendStr(e, '\n%s' % s) else: except: '%s: ERROR IN PRINTING' % f.__name__) raise _exceptionLogged.__doc__ = f.__doc__ return _exceptionLogged return _decoratorFunc
# GoldenRatio = (1. + math.sqrt(5.)) / 2.
[docs]class GoldenRectangle:
[docs] @staticmethod def getLongerEdge(shorter): return shorter * GoldenRatio
[docs] @staticmethod def getShorterEdge(longer): return longer / GoldenRatio
[docs]def nullGen(): # generator that ends immediately if False: # yield that never runs but still exists, making this func a generator yield None
[docs]def loopGen(l): # generator that yields the items of an iterable object forever def _gen(l): while True: for item in l: yield item gen = _gen(l) # don't leak _gen = None return gen
[docs]def makeFlywheelGen(objects, countList=None, countFunc=None, scale=None): # iterates and finally yields a flywheel generator object # the number of appearances for each object is controlled by passing in # a list of counts, or a functor that returns a count when called with # an object from the 'objects' list. # if scale is provided, all counts are scaled by the scale value and then int()'ed. def flywheel(index2objectAndCount): # generator to produce a sequence whose elements appear a specific number of times while len(index2objectAndCount) > 0: keyList = list(index2objectAndCount.keys()) for key in keyList: if index2objectAndCount[key][1] > 0: yield index2objectAndCount[key][0] index2objectAndCount[key][1] -= 1 if index2objectAndCount[key][1] <= 0: del index2objectAndCount[key] # if we were not given a list of counts, create it by calling countFunc if countList is None: countList = [] for object in objects: yield None countList.append(countFunc(object)) if scale is not None: # scale the counts if we've got a scale factor for i in range(len(countList)): yield None if countList[i] > 0: countList[i] = max(1, int(countList[i] * scale)) # create a dict for the flywheel to use during its iteration to efficiently select # the objects for the sequence index2objectAndCount = {} for i in range(len(countList)): yield None index2objectAndCount[i] = [objects[i], countList[i]] # create the flywheel generator yield flywheel(index2objectAndCount)
[docs]def flywheel(*args, **kArgs): # create a flywheel generator # see arguments and comments in flywheelGen above # example usage: """ >>> for i in flywheel([1,2,3], countList=[10, 5, 1]): ... print i, ... 1 2 3 1 2 1 2 1 2 1 2 1 1 1 1 1 """ for flywheel in makeFlywheelGen(*args, **kArgs): pass return flywheel
if __debug__:
[docs] def quickProfile(name="unnamed"): import pstats def profileDecorator(f): if not ConfigVariableBool("use-profiler", False): return f def _profiled(*args, **kArgs): # must do this in here because we don't have base/simbase # at the time that PythonUtil is loaded if not ConfigVariableBool("profile-debug", False): #dumb timings clock = ClockObject.getGlobalClock() st = clock.getRealTime() f(*args, **kArgs) s = clock.getRealTime() - st print("Function %s.%s took %s seconds"%(f.__module__, f.__name__,s)) else: import profile as prof, pstats #detailed profile, stored in base.stats under ( if not hasattr(base, "stats"): base.stats = {} if not base.stats.get(name): base.stats[name] = [] prof.runctx('f(*args, **kArgs)', {'f':f,'args':args,'kArgs':kArgs},None,"") s=pstats.Stats("") #p=hotshot.Profile("") #p.runctx('f(*args, **kArgs)', {'f':f,'args':args,'kArgs':kArgs},None) #s = hotshot.stats.load("") s.strip_dirs() s.sort_stats("cumulative") base.stats[name].append(s) _profiled.__doc__ = f.__doc__ return _profiled return profileDecorator
[docs]def getTotalAnnounceTime(): td=0 for objs in base.stats.values(): for stat in objs: td+=getAnnounceGenerateTime(stat) return td
[docs]def getAnnounceGenerateTime(stat): val=0 stats=stat.stats for i in list(stats.keys()): if i[2] == "announceGenerate": newVal = stats[i][3] if newVal > val: val = newVal return val
[docs]class MiniLog:
[docs] def __init__(self, name): self.indent = 1 = name self.lines = []
def __str__(self): return '%s\nMiniLog: %s\n%s\n%s\n%s' % \ ('*'*50,, '-'*50, '\n'.join(self.lines), '*'*50)
[docs] def enterFunction(self, funcName, *args, **kw): rArgs = [repr(x)+', ' for x in args] + \ [ x + ' = ' + '%s, ' % repr(y) for x,y in kw.items()] if not rArgs: rArgs = '()' else: rArgs = '(' + functools.reduce(str.__add__,rArgs)[:-2] + ')' line = '%s%s' % (funcName, rArgs) self.appendFunctionCall(line) self.indent += 1 return line
[docs] def exitFunction(self): self.indent -= 1 return self.indent
[docs] def appendFunctionCall(self, line): self.lines.append(' '*(self.indent*2) + line) return line
[docs] def appendLine(self, line): self.lines.append(' '*(self.indent*2) + '<< ' + line + ' >>') return line
[docs] def flush(self): outStr = str(self) self.indent = 0 self.lines = [] return outStr
[docs]class MiniLogSentry:
[docs] def __init__(self, log, funcName, *args, **kw): self.log = log if self.log: self.log.enterFunction(funcName, *args, **kw)
def __del__(self): if self.log: self.log.exitFunction() del self.log
[docs]def logBlock(id, msg): print('<< LOGBLOCK(%03d)' % id) print(str(msg)) print('/LOGBLOCK(%03d) >>' % id)
[docs]class HierarchyException(Exception): JOSWILSO = 0
[docs] def __init__(self, owner, description): self.owner = owner self.desc = description
def __str__(self): return '(%s): %s' % (self.owner, self.desc) def __repr__(self): return 'HierarchyException(%s)' % (self.owner, )
[docs]def formatTimeCompact(seconds): # returns string in format '1d3h22m43s' result = '' a = int(seconds) seconds = a % 60 a //= 60 if a > 0: minutes = a % 60 a //= 60 if a > 0: hours = a % 24 a //= 24 if a > 0: days = a result += '%sd' % days result += '%sh' % hours result += '%sm' % minutes result += '%ss' % seconds return result
if __debug__ and __name__ == '__main__': ftc = formatTimeCompact assert ftc(0) == '0s' assert ftc(1) == '1s' assert ftc(60) == '1m0s' assert ftc(64) == '1m4s' assert ftc(60*60) == '1h0m0s' assert ftc(24*60*60) == '1d0h0m0s' assert ftc(24*60*60 + 2*60*60 + 34*60 + 12) == '1d2h34m12s' del ftc
[docs]def formatTimeExact(seconds): # like formatTimeCompact but leaves off '0 seconds', '0 minutes' etc. for # times that are e.g. 1 hour, 3 days etc. # returns string in format '1d3h22m43s' result = '' a = int(seconds) seconds = a % 60 a //= 60 if a > 0: minutes = a % 60 a //= 60 if a > 0: hours = a % 24 a //= 24 if a > 0: days = a result += '%sd' % days if hours or minutes or seconds: result += '%sh' % hours if minutes or seconds: result += '%sm' % minutes if seconds or result == '': result += '%ss' % seconds return result
if __debug__ and __name__ == '__main__': fte = formatTimeExact assert fte(0) == '0s' assert fte(1) == '1s' assert fte(2) == '2s' assert fte(61) == '1m1s' assert fte(60) == '1m' assert fte(60*60) == '1h' assert fte(24*60*60) == '1d' assert fte((24*60*60) + (2 * 60)) == '1d0h2m' del fte
[docs]class AlphabetCounter: # object that produces 'A', 'B', 'C', ... 'AA', 'AB', etc.
[docs] def __init__(self): self._curCounter = ['A']
[docs] def next(self): result = ''.join([c for c in self._curCounter]) index = -1 while True: curChar = self._curCounter[index] if curChar == 'Z': nextChar = 'A' carry = True else: nextChar = chr(ord(self._curCounter[index])+1) carry = False self._curCounter[index] = nextChar if carry: if (-index) == len(self._curCounter): self._curCounter = ['A',] + self._curCounter break else: index -= 1 carry = False else: break return result
__next__ = next
if __debug__ and __name__ == '__main__': def testAlphabetCounter(): tempList = [] ac = AlphabetCounter() for i in range(26*3): tempList.append( assert tempList == [ 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', 'AA','AB','AC','AD','AE','AF','AG','AH','AI','AJ','AK','AL','AM','AN','AO','AP','AQ','AR','AS','AT','AU','AV','AW','AX','AY','AZ', 'BA','BB','BC','BD','BE','BF','BG','BH','BI','BJ','BK','BL','BM','BN','BO','BP','BQ','BR','BS','BT','BU','BV','BW','BX','BY','BZ',] ac = AlphabetCounter() num = 26 # A-Z num += (26*26) # AA-ZZ num += 26 # AAZ num += 1 # ABA num += 2 # ABC for i in range(num): x = assert x == 'ABC' testAlphabetCounter() del testAlphabetCounter
[docs]class Default: # represents 'use the default value' # useful for keyword arguments to virtual methods pass
superLogFile = None
[docs]def startSuperLog(customFunction = None): global superLogFile if not superLogFile: superLogFile = open("c:\\temp\\superLog.txt", "w") def trace_dispatch(a,b,c): if b == 'call' and a.f_code.co_name != '?' and a.f_code.co_name.find("safeRepr") < 0: vars = dict(a.f_locals) if 'self' in vars: del vars['self'] if '__builtins__' in vars: del vars['__builtins__'] for i in vars: vars[i] = safeReprTypeOnFail(vars[i]) if customFunction: superLogFile.write( "before = %s\n"%customFunction()) superLogFile.write( "%s(%s):%s:%s\n"%(a.f_code.co_filename.split("\\")[-1],a.f_code.co_firstlineno, a.f_code.co_name, vars)) if customFunction: superLogFile.write( "after = %s\n"%customFunction()) return trace_dispatch sys.settrace(trace_dispatch)
[docs]def endSuperLog(): global superLogFile if superLogFile: sys.settrace(None) superLogFile.close() superLogFile = None
[docs]def configIsToday(configName): # TODO: replace usage of strptime with something else # returns true if config string is a valid representation of today's date today = time.localtime() confStr = config.GetString(configName, '') for format in ('%m/%d/%Y', '%m-%d-%Y', '%m.%d.%Y'): try: confDate = time.strptime(confStr, format) except ValueError: pass else: if (confDate.tm_year == today.tm_year and confDate.tm_mon == today.tm_mon and confDate.tm_mday == today.tm_mday): return True return False
[docs]def typeName(o): if hasattr(o, '__class__'): return o.__class__.__name__ else: return o.__name__
[docs]def safeTypeName(o): try: return typeName(o) except: pass try: return type(o) except: pass return '<failed safeTypeName()>'
[docs]def histogramDict(l): d = {} for e in l: d.setdefault(e, 0) d[e] += 1 return d
[docs]def unescapeHtmlString(s): # converts %## to corresponding character # replaces '+' with ' ' result = '' i = 0 while i < len(s): char = s[i] if char == '+': char = ' ' elif char == '%': if i < (len(s)-2): num = int(s[i+1:i+3], 16) char = chr(num) i += 2 i += 1 result += char return result
[docs]class PriorityCallbacks: """ manage a set of prioritized callbacks, and allow them to be invoked in order of priority """
[docs] def __init__(self): self._callbacks = []
[docs] def clear(self): del self._callbacks[:]
[docs] def add(self, callback, priority=None): if priority is None: priority = 0 callbacks = self._callbacks lo = 0 hi = len(callbacks) while lo < hi: mid = (lo + hi) // 2 if priority < callbacks[mid][0]: hi = mid else: lo = mid + 1 item = (priority, callback) callbacks.insert(lo, item) return item
[docs] def remove(self, item): self._callbacks.remove(item)
def __call__(self): for priority, callback in self._callbacks: callback()
builtins.Functor = Functor builtins.Stack = Stack builtins.Queue = Queue builtins.SerialNumGen = SerialNumGen builtins.SerialMaskedGen = SerialMaskedGen builtins.ScratchPad = ScratchPad builtins.uniqueName = uniqueName builtins.serialNum = serialNum if __debug__: builtins.profiled = profiled builtins.exceptionLogged = exceptionLogged builtins.itype = itype builtins.appendStr = appendStr builtins.bound = bound builtins.clamp = clamp builtins.lerp = lerp builtins.makeList = makeList builtins.makeTuple = makeTuple if __debug__: builtins.printStack = printStack builtins.printReverseStack = printReverseStack builtins.printVerboseStack = printVerboseStack builtins.DelayedCall = DelayedCall builtins.DelayedFunctor = DelayedFunctor builtins.FrameDelayedCall = FrameDelayedCall builtins.SubframeCall = SubframeCall builtins.invertDict = invertDict builtins.invertDictLossless = invertDictLossless builtins.getBase = getBase builtins.getRepository = getRepository builtins.safeRepr = safeRepr builtins.fastRepr = fastRepr builtins.nullGen = nullGen builtins.flywheel = flywheel builtins.loopGen = loopGen if __debug__: builtins.StackTrace = StackTrace = report builtins.pstatcollect = pstatcollect builtins.MiniLog = MiniLog builtins.MiniLogSentry = MiniLogSentry builtins.logBlock = logBlock builtins.HierarchyException = HierarchyException builtins.deeptype = deeptype builtins.Default = Default builtins.configIsToday = configIsToday builtins.typeName = typeName builtins.safeTypeName = safeTypeName builtins.histogramDict = histogramDict