Source code for direct.interval.MetaInterval

"""
This module defines the various "meta intervals", which execute other
intervals either in parallel or in a specified sequential order.
"""

__all__ = ['MetaInterval', 'Sequence', 'Parallel', 'ParallelEndTogether', 'Track']

from panda3d.core import *
from panda3d.direct import *
from direct.directnotify.DirectNotifyGlobal import *
from .IntervalManager import ivalMgr
from . import Interval
from direct.task.Task import TaskManager
#if __debug__:
#    import direct.showbase.PythonUtil as PythonUtil

PREVIOUS_END = CMetaInterval.RSPreviousEnd
PREVIOUS_START = CMetaInterval.RSPreviousBegin
TRACK_START = CMetaInterval.RSLevelBegin

[docs]class MetaInterval(CMetaInterval): # This is a Python-C++ hybrid class. MetaInterval is a Python # extension of the C++ class CMetaInterval, which adds some # Python-specific features (like list management). # This is the base class of Sequence, Parallel, and Track. notify = directNotify.newCategory("MetaInterval") SequenceNum = 1
[docs] def __init__(self, *ivals, **kw): #if __debug__: # self.debugInitTraceback = PythonUtil.StackTrace( # "create interval", 1, 10) name = None #if len(ivals) == 2 and isinstance(ivals[1], str): # # If the second parameter is a string, it's the name. # name = ivals[1] # ivals = ivals[0] #else: # Look for the name in the keyword params. if 'name' in kw: name = kw['name'] del kw['name'] # If the keyword "autoPause" or "autoFinish" is defined to # non-zero, it means the interval may be automatically paused # or finished when CIntervalManager::interrupt() is called. # This is generally called only on a catastrophic situation # (for instance, the connection to the server being lost) when # we have to exit right away; these keywords indicate # intervals that might not be cleaned up by their owners. autoPause = 0 autoFinish = 0 if 'autoPause' in kw: autoPause = kw['autoPause'] del kw['autoPause'] if 'autoFinish' in kw: autoFinish = kw['autoFinish'] del kw['autoFinish'] # A duration keyword specifies the duration the interval will # appear to have for the purposes of computing the start time # for subsequent intervals in a sequence or track. self.phonyDuration = -1 if 'duration' in kw: self.phonyDuration = kw['duration'] del kw['duration'] if kw: self.notify.error("Unexpected keyword parameters: %s" % (list(kw.keys()))) # We must allow the old style: Track([ival0, ival1, ...]) as # well as the new style: Track(ival0, ival1, ...) # Note: this breaks in the case of a Track with one tuple: # Track((0, ival0),). We could go through some effort to fix # this case, but for now I prefer just to document it as a # bug, since it will go away when we eventually remove support # for the old interface. #if len(ivals) == 1 and \ # (isinstance(ivals[0], tuple) or \ # isinstance(ivals[0], list)): # self.ivals = ivals[0] #else: self.ivals = ivals self.__ivalsDirty = 1 if name is None: name = self.__class__.__name__ + '-%d' if '%' in name: name = name % (self.SequenceNum) MetaInterval.SequenceNum += 1 CMetaInterval.__init__(self, name) self.__manager = ivalMgr self.setAutoPause(autoPause) self.setAutoFinish(autoFinish) self.pstats = None if __debug__ and TaskManager.taskTimerVerbose: self.pname = name.split('-', 1)[0] self.pstats = PStatCollector("App:Tasks:ivalLoop:%s" % (self.pname)) self.pythonIvals = [] # If we are running in debug mode, we validate the intervals # in the list right away. There's no good reason to do this, # except that it makes it easier for the programmer to detect # when a MetaInterval is misdefined at creation time. assert self.validateComponents(self.ivals)
# Functions to make the MetaInterval object act just like a Python # list of intervals:
[docs] def append(self, ival): # Appends a single interval to the list so far. if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) self.ivals.append(ival) self.__ivalsDirty = 1 assert self.validateComponent(ival)
[docs] def extend(self, ivals): # Appends a list of intervals to the list so far. self += ivals
[docs] def count(self, ival): # Returns the number of occurrences of the indicated interval. return self.ivals.count(ival)
[docs] def index(self, ival): # Returns the position of the indicated interval within the list. return self.ivals.index(ival)
[docs] def insert(self, index, ival): # Inserts the given interval into the middle of the list. if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) self.ivals.insert(index, ival) self.__ivalsDirty = 1 assert self.validateComponent(ival)
[docs] def pop(self, index = None): # Returns element index (or the last element) and removes it # from the list. if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) self.__ivalsDirty = 1 if index is None: return self.ivals.pop() else: return self.ivals.pop(index)
[docs] def remove(self, ival): # Removes the indicated interval from the list. if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) self.ivals.remove(ival) self.__ivalsDirty = 1
[docs] def reverse(self): # Reverses the order of the intervals. if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) self.ivals.reverse() self.__ivalsDirty = 1
[docs] def sort(self, cmpfunc = None): # Sorts the intervals. (?) if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) self.__ivalsDirty = 1 if cmpfunc is None: self.ivals.sort() else: self.ivals.sort(cmpfunc)
def __len__(self): return len(self.ivals) def __getitem__(self, index): return self.ivals[index] def __setitem__(self, index, value): if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) self.ivals[index] = value self.__ivalsDirty = 1 assert self.validateComponent(value) def __delitem__(self, index): if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) del self.ivals[index] self.__ivalsDirty = 1 def __getslice__(self, i, j): if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) return self.__class__(self.ivals[i: j]) def __setslice__(self, i, j, s): if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) self.ivals[i: j] = s self.__ivalsDirty = 1 assert self.validateComponents(s) def __delslice__(self, i, j): if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) del self.ivals[i: j] self.__ivalsDirty = 1 def __iadd__(self, other): if isinstance(self.ivals, tuple): self.ivals = list(self.ivals) if isinstance(other, MetaInterval): assert self.__class__ == other.__class__ ivals = other.ivals else: ivals = list(other) self.ivals += ivals self.__ivalsDirty = 1 assert self.validateComponents(ivals) return self def __add__(self, other): copy = self[:] copy += other return copy # Functions to define sequence, parallel, and track behaviors:
[docs] def addSequence(self, list, name, relTime, relTo, duration): # Adds the given list of intervals to the MetaInterval to be # played one after the other. self.pushLevel(name, relTime, relTo) for ival in list: self.addInterval(ival, 0.0, PREVIOUS_END) self.popLevel(duration)
[docs] def addParallel(self, list, name, relTime, relTo, duration): # Adds the given list of intervals to the MetaInterval to be # played simultaneously; all will start at the same time. self.pushLevel(name, relTime, relTo) for ival in list: self.addInterval(ival, 0.0, TRACK_START) self.popLevel(duration)
[docs] def addParallelEndTogether(self, list, name, relTime, relTo, duration): # Adds the given list of intervals to the MetaInterval to be # played simultaneously; all will end at the same time, but # the longest interval will be started first to achieve this. maxDuration = 0 for ival in list: maxDuration = max(maxDuration, ival.getDuration()) self.pushLevel(name, relTime, relTo) for ival in list: self.addInterval(ival, maxDuration - ival.getDuration(), TRACK_START) self.popLevel(duration)
[docs] def addTrack(self, trackList, name, relTime, relTo, duration): # Adds a "track list". This is a list of tuples of the form: # # (<delay>, <Interval>, # PREVIOUS_END | PREVIOUS_START | TRACK_START) # # where <delay> is a relative time, in seconds, for the # <Interval> to start, relative to either the end of the # previous interval (PREVIOUS_END), the start of the previous # interval (PREVIOUS_START) or the start of the track list # (TRACK_START). If the relative code is omitted, the default # is TRACK_START. self.pushLevel(name, relTime, relTo) for tupleObj in trackList: if isinstance(tupleObj, tuple) or \ isinstance(tupleObj, list): relTime = tupleObj[0] ival = tupleObj[1] if len(tupleObj) >= 3: relTo = tupleObj[2] else: relTo = TRACK_START self.addInterval(ival, relTime, relTo) else: self.notify.error("Not a tuple in Track: %s" % (tupleObj,)) self.popLevel(duration)
[docs] def addInterval(self, ival, relTime, relTo): # Adds the given interval to the MetaInterval. if isinstance(ival, CInterval): # It's a C++-style Interval, so add it directly. if getattr(ival, "inPython", 0): # Actually, it's been flagged to run in Python, even # though it's a C++ Interval. It's probably got some # Python functors that must be invoked at runtime to # define some of its parameters. Treat it as a Python # interval. index = len(self.pythonIvals) self.pythonIvals.append(ival) self.addExtIndex(index, ival.getName(), ival.getDuration(), ival.getOpenEnded(), relTime, relTo) elif isinstance(ival, MetaInterval): # It's another MetaInterval, so copy in its intervals # directly to this object. We could just store the # MetaInterval itself, which would work, but we get a # performance advantage by flattening out the deeply # nested hierarchy into a linear list within the root # CMetaInterval object. ival.applyIvals(self, relTime, relTo) else: # Nope, a perfectly ordinary C++ interval. Hooray! self.addCInterval(ival, relTime, relTo) elif isinstance(ival, Interval.Interval): # It's a Python-style Interval, so add it as an external. index = len(self.pythonIvals) self.pythonIvals.append(ival) if self.pstats: ival.pstats = PStatCollector(self.pstats, ival.pname) self.addExtIndex(index, ival.getName(), ival.getDuration(), ival.getOpenEnded(), relTime, relTo) else: self.notify.error("Not an Interval: %s" % (ival,))
# Functions to support automatic playback of MetaIntervals along # with all of their associated Python callbacks:
[docs] def setManager(self, manager): self.__manager = manager CMetaInterval.setManager(self, manager)
[docs] def getManager(self): return self.__manager
manager = property(getManager, setManager)
[docs] def setT(self, t): self.__updateIvals() CMetaInterval.setT(self, t)
t = property(CMetaInterval.getT, setT)
[docs] def start(self, startT = 0.0, endT = -1.0, playRate = 1.0): self.__updateIvals() self.setupPlay(startT, endT, playRate, 0) self.__manager.addInterval(self)
[docs] def loop(self, startT = 0.0, endT = -1.0, playRate = 1.0): self.__updateIvals() self.setupPlay(startT, endT, playRate, 1) self.__manager.addInterval(self)
[docs] def pause(self): if self.getState() == CInterval.SStarted: self.privInterrupt() self.__manager.removeInterval(self) self.privPostEvent() return self.getT()
[docs] def resume(self, startT = None): self.__updateIvals() if startT is not None: self.setT(startT) self.setupResume() self.__manager.addInterval(self)
[docs] def resumeUntil(self, endT): self.__updateIvals() self.setupResumeUntil(endT) self.__manager.addInterval(self)
[docs] def finish(self): self.__updateIvals() state = self.getState() if state == CInterval.SInitial: self.privInstant() elif state != CInterval.SFinal: self.privFinalize() self.__manager.removeInterval(self) self.privPostEvent()
[docs] def clearToInitial(self): # This is overloaded at the Python level to properly call # pause() at the Python level, then upcall to finish the job # at the C++ level. self.pause() CMetaInterval.clearToInitial(self)
# Internal functions:
[docs] def validateComponent(self, component): # This is called only in debug mode to verify that the # indicated component added to the MetaInterval is appropriate # to this type of MetaInterval. In most cases except Track, # this is the same as asking that the component is itself an # Interval. return isinstance(component, CInterval) or \ isinstance(component, Interval.Interval)
[docs] def validateComponents(self, components): # This is called only in debug mode to verify that all the # components on the indicated list are appropriate to this # type of MetaInterval. for component in components: if not self.validateComponent(component): return 0 return 1
def __updateIvals(self): # The MetaInterval object does not create the C++ list of # Intervals immediately; rather, it stores a Python list of # Intervals that will be compiled into the C++ list the first # time it is needed. # This design allows us to avoid creation of the C++ list for # nested MetaInterval objects, instead copying all nested # MetaInterval hierarchy into the root CMetaInterval object, # for a performance benefit. # This function is called only on the root MetaInterval # object, when it is time to build the C++ list for itself. if self.__ivalsDirty: self.clearIntervals() self.applyIvals(self, 0, TRACK_START) self.__ivalsDirty = 0
[docs] def clearIntervals(self): # This overrides the function defined at the C++ level to # reset the inPython flag. Clearing out the intervals list # allows us to run entirely in C++ again, at least until a new # Python interval gets added. CMetaInterval.clearIntervals(self) self.inPython = 0
[docs] def applyIvals(self, meta, relTime, relTo): # Add the intervals listed in this object to the given # MetaInterval object at the C++ level. This will make the # other MetaInterval object ready to play the intervals. # This function should be overridden in a derived class to # change the intepretation of the intervals in this list. In # the case of a MetaInterval directly, this is valid only if # the list has only zero or one intervals. if len(self.ivals) == 0: pass elif len(self.ivals) == 1: meta.addInterval(self.ivals[0], relTime, relTo) else: self.notify.error("Cannot build list from MetaInterval directly.")
[docs] def setPlayRate(self, playRate): """ Changes the play rate of the interval. If the interval is already started, this changes its speed on-the-fly. Note that since playRate is a parameter to start() and loop(), the next call to start() or loop() will reset this parameter. """ if self.isPlaying(): self.pause() CMetaInterval.setPlayRate(self, playRate) self.resume() else: CMetaInterval.setPlayRate(self, playRate)
play_rate = property(CMetaInterval.getPlayRate, setPlayRate) def __doPythonCallbacks(self): # This function invokes any Python-level Intervals that need # to be invoked at this point in time. It must be called # after any call to setT() or setFinalT() or stepPlay(), or # some such; basically any function that might invoke an # interval. The C++ base class will invoke whatever C++ # intervals it can, and then indicate the Python intervals # that must be invoked through this interface. ival = None try: while self.isEventReady(): index = self.getEventIndex() t = self.getEventT() eventType = self.getEventType() self.popEvent() ival = self.pythonIvals[index] ival.privDoEvent(t, eventType) ival.privPostEvent() ival = None except: if ival is not None: print("Exception occurred while processing %s of %s:" % (ival.getName(), self.getName())) else: print("Exception occurred while processing %s:" % (self.getName())) print(self) raise
[docs] def privDoEvent(self, t, event): # This function overrides the C++ function to initialize the # intervals first if necessary. if self.pstats: self.pstats.start() self.__updateIvals() CMetaInterval.privDoEvent(self, t, event) if self.pstats: self.pstats.stop()
[docs] def privPostEvent(self): if self.pstats: self.pstats.start() self.__doPythonCallbacks() CMetaInterval.privPostEvent(self) if self.pstats: self.pstats.stop()
[docs] def setIntervalStartTime(self, *args, **kw): # This function overrides from the parent level to force it to # update the interval list first, if necessary. self.__updateIvals() # Once we have monkeyed with the interval timings, we'd better # run the whole thing as a monolithic Python interval, since # we can't extract the ivals list back out and append them # into a parent MetaInterval. self.inPython = 1 return CMetaInterval.setIntervalStartTime(self, *args, **kw)
[docs] def getIntervalStartTime(self, *args, **kw): # This function overrides from the parent level to force it to # update the interval list first, if necessary. self.__updateIvals() return CMetaInterval.getIntervalStartTime(self, *args, **kw)
[docs] def getDuration(self): # This function overrides from the parent level to force it to # update the interval list first, if necessary. self.__updateIvals() return CMetaInterval.getDuration(self)
duration = property(getDuration) def __repr__(self, *args, **kw): # This function overrides from the parent level to force it to # update the interval list first, if necessary. self.__updateIvals() return CMetaInterval.__repr__(self, *args, **kw) def __str__(self, *args, **kw): # This function overrides from the parent level to force it to # update the interval list first, if necessary. self.__updateIvals() return CMetaInterval.__str__(self, *args, **kw)
[docs] def timeline(self, out = None): # This function overrides from the parent level to force it to # update the interval list first, if necessary. self.__updateIvals() if out is None: out = ostream CMetaInterval.timeline(self, out)
add_sequence = addSequence add_parallel = addParallel add_parallel_end_together = addParallelEndTogether add_track = addTrack add_interval = addInterval set_manager = setManager get_manager = getManager set_t = setT resume_until = resumeUntil clear_to_initial = clearToInitial clear_intervals = clearIntervals set_play_rate = setPlayRate priv_do_event = privDoEvent priv_post_event = privPostEvent set_interval_start_time = setIntervalStartTime get_interval_start_time = getIntervalStartTime get_duration = getDuration
[docs]class Sequence(MetaInterval):
[docs] def applyIvals(self, meta, relTime, relTo): meta.addSequence(self.ivals, self.getName(), relTime, relTo, self.phonyDuration)
[docs]class Parallel(MetaInterval):
[docs] def applyIvals(self, meta, relTime, relTo): meta.addParallel(self.ivals, self.getName(), relTime, relTo, self.phonyDuration)
[docs]class ParallelEndTogether(MetaInterval):
[docs] def applyIvals(self, meta, relTime, relTo): meta.addParallelEndTogether(self.ivals, self.getName(), relTime, relTo, self.phonyDuration)
[docs]class Track(MetaInterval):
[docs] def applyIvals(self, meta, relTime, relTo): meta.addTrack(self.ivals, self.getName(), relTime, relTo, self.phonyDuration)
[docs] def validateComponent(self, tupleObj): # This is called only in debug mode to verify that the # indicated component added to the MetaInterval is appropriate # to this type of MetaInterval. In most cases except Track, # this is the same as asking that the component is itself an # Interval. if not (isinstance(tupleObj, tuple) or \ isinstance(tupleObj, list)): # It's not a tuple. return 0 relTime = tupleObj[0] ival = tupleObj[1] if len(tupleObj) >= 3: relTo = tupleObj[2] else: relTo = TRACK_START if not (isinstance(relTime, float) or \ isinstance(relTime, int)): # First parameter is not a number. return 0 if not MetaInterval.validateComponent(self, ival): # Second parameter is not an interval. return 0 if relTo != PREVIOUS_END and \ relTo != PREVIOUS_START and \ relTo != TRACK_START: # Third parameter is an invalid value. return 0 # Looks good. return 1