tessl/pypi-panda3d
Panda3D is a framework for 3D rendering and game development for Python and C++ programs.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
task-event.mddocs/
Task and Event Management
Panda3D provides a comprehensive system for frame-based task scheduling and event-driven programming, essential for game logic, animations, and inter-object communication.
Capabilities
TaskManager - Frame-Based Task Scheduling
The TaskManager handles frame-based tasks that run continuously or periodically, forming the backbone of game logic execution.
class TaskManager:
def add(self,
task: callable,
name: str,
priority: int = 0,
uponDeath: callable = None,
appendTask: bool = False,
owner: object = None) -> Task:
"""
Add a task to run every frame.
Args:
task: Function to call each frame, must return task.cont or task.done
name: Unique name for the task
priority: Task priority (lower numbers run first)
uponDeath: Function to call when task completes
appendTask: Whether to append Task object as last argument
owner: Object that owns this task
Returns:
Task object for management
"""
def remove(self, taskOrName: str | Task) -> bool:
"""Remove a task by name or Task object."""
def removeTasksMatching(self, pattern: str) -> int:
"""Remove all tasks with names matching glob pattern."""
def step(self) -> None:
"""Execute one frame of all tasks."""
def run(self) -> None:
"""Run task manager main loop (blocking)."""
def stop(self) -> None:
"""Stop the task manager main loop."""
def hasTaskNamed(self, name: str) -> bool:
"""Check if task with given name exists."""
def getTasksNamed(self, name: str) -> List[Task]:
"""Get all tasks with given name."""
def getAllTasks(self) -> List[Task]:
"""Get list of all active tasks."""
def getTasks(self) -> List[Task]:
"""Get list of all tasks."""
def getNumTasks(self) -> int:
"""Get total number of active tasks."""
def doMethodLater(self,
delayTime: float,
task: callable,
name: str,
extraArgs: List = [],
priority: int = 0,
uponDeath: callable = None,
owner: object = None) -> Task:
"""Schedule a task to run after a delay."""
def doMethodLaterPriority(self,
delayTime: float,
task: callable,
name: str,
priority: int,
extraArgs: List = []) -> Task:
"""Schedule a delayed task with specific priority."""Task Objects and Control
Individual task objects provide fine-grained control over task execution and state.
class Task:
# Task return constants
cont: int # Continue running task next frame
done: int # Task completed, remove from manager
again: int # Restart task from beginning
def __init__(self, function: callable = None, name: str = None) -> None:
"""Create new task object."""
def setFunction(self, function: callable) -> None:
"""Set task function."""
def getFunction(self) -> callable:
"""Get task function."""
def setName(self, name: str) -> None:
"""Set task name."""
def getName(self) -> str:
"""Get task name."""
def setPriority(self, priority: int) -> None:
"""Set task priority."""
def getPriority(self) -> int:
"""Get task priority."""
def setDelay(self, delay: float) -> None:
"""Set delay before task starts."""
def getDelay(self) -> float:
"""Get task delay."""
def setUponDeath(self, function: callable) -> None:
"""Set function to call when task completes."""
def getUponDeath(self) -> callable:
"""Get upon-death function."""
def remove(self) -> None:
"""Remove this task from manager."""
def pause(self) -> None:
"""Pause task execution."""
def resume(self) -> None:
"""Resume paused task."""
def isPaused(self) -> bool:
"""Check if task is paused."""
def getElapsedTime(self) -> float:
"""Get time elapsed since task started."""
def getElapsedFrames(self) -> int:
"""Get frames elapsed since task started."""
def getWakeTime(self) -> float:
"""Get time when delayed task will wake up."""
# Task state properties
time: float # Current task time
frame: int # Current frame count
dt: float # Delta time since last frame
id: int # Unique task IDMessenger - Event System
The Messenger provides a global publish-subscribe event system for inter-object communication.
class Messenger:
def accept(self,
event: str,
method: callable,
extraArgs: List = [],
persistent: int = 1) -> None:
"""
Accept an event and bind it to a method.
Args:
event: Event name to listen for
method: Function to call when event occurs
extraArgs: Additional arguments to pass to method
persistent: How many times to accept event (0 = forever)
"""
def acceptOnce(self,
event: str,
method: callable,
extraArgs: List = []) -> None:
"""Accept an event only once."""
def ignore(self, event: str) -> None:
"""Stop accepting a specific event."""
def ignoreAll(self) -> None:
"""Stop accepting all events."""
def isAccepting(self, event: str) -> bool:
"""Check if currently accepting an event."""
def getAllAccepting(self) -> List[str]:
"""Get list of all accepted events."""
def send(self, event: str, sentArgs: List = []) -> None:
"""Send an event to all listeners."""
def clear(self) -> None:
"""Clear all event handlers."""
def isEmpty(self) -> bool:
"""Check if messenger has no handlers."""
def replaceMethod(self,
oldMethod: callable,
newMethod: callable) -> int:
"""Replace method in all event handlers."""
def toggleVerbose(self) -> None:
"""Toggle verbose event logging."""
def setVerbose(self, verbose: bool) -> None:
"""Enable or disable verbose event logging."""DirectObject - Event-Aware Base Class
DirectObject provides a convenient base class that integrates with the event system.
class DirectObject:
def __init__(self) -> None:
"""Initialize DirectObject with event handling."""
def accept(self,
event: str,
method: callable,
extraArgs: List = []) -> None:
"""Accept event (automatically cleaned up on destruction)."""
def acceptOnce(self,
event: str,
method: callable,
extraArgs: List = []) -> None:
"""Accept event once."""
def ignore(self, event: str) -> None:
"""Ignore specific event."""
def ignoreAll(self) -> None:
"""Ignore all events for this object."""
def isAccepting(self, event: str) -> bool:
"""Check if accepting event."""
def getAllAccepting(self) -> List[str]:
"""Get all accepted events."""
def destroy(self) -> None:
"""Clean up object and remove all event handlers."""
def addTask(self,
task: callable,
name: str,
priority: int = 0,
uponDeath: callable = None,
appendTask: bool = False) -> Task:
"""Add task (automatically cleaned up on destruction)."""
def doMethodLater(self,
delayTime: float,
task: callable,
name: str,
extraArgs: List = []) -> Task:
"""Schedule delayed task."""
def removeTask(self, taskOrName: str | Task) -> None:
"""Remove specific task."""
def removeAllTasks(self) -> None:
"""Remove all tasks for this object."""Common Event Types
Standard events generated by Panda3D systems.
# Window events
"window-event" # Window state changes
"window-close-request" # User requests window close
# Input events
"escape" # Escape key pressed
"space" # Space key pressed
"enter" # Enter key pressed
"mouse1" # Left mouse button pressed
"mouse1-up" # Left mouse button released
"mouse2" # Middle mouse button pressed
"mouse2-up" # Middle mouse button released
"mouse3" # Right mouse button pressed
"mouse3-up" # Right mouse button released
"wheel_up" # Mouse wheel up
"wheel_down" # Mouse wheel down
# Key events (format: key name, key-up)
"a", "a-up" # A key pressed/released
"shift", "shift-up" # Shift key pressed/released
"control", "control-up" # Control key pressed/released
"alt", "alt-up" # Alt key pressed/released
# Arrow keys
"arrow_up", "arrow_up-up"
"arrow_down", "arrow_down-up"
"arrow_left", "arrow_left-up"
"arrow_right", "arrow_right-up"
# Function keys
"f1", "f1-up" # F1 through F12
# ... f2 through f12
# Task events
"task-added" # Task added to manager
"task-removed" # Task removed from managerClock and Timing
Access global timing information for frame-based calculations.
class ClockObject:
def getDt(self) -> float:
"""Get delta time since last frame."""
def getFrameTime(self) -> float:
"""Get time of current frame."""
def getRealTime(self) -> float:
"""Get real elapsed time since start."""
def getFrameCount(self) -> int:
"""Get total frame count."""
def getAverageFrameRate(self) -> float:
"""Get average frame rate."""
def getMaxDt(self) -> float:
"""Get maximum allowed delta time."""
def setMaxDt(self, maxDt: float) -> None:
"""Set maximum delta time limit."""
def tick(self) -> None:
"""Advance clock by one frame."""
# Global clock instance
globalClock: ClockObjectUsage Examples
Basic Task Management
from direct.showbase.ShowBase import ShowBase
from direct.task import Task
from panda3d.core import Vec3
class TaskDemo(ShowBase):
def __init__(self):
ShowBase.__init__(self)
# Create object to animate
self.cube = self.loader.loadModel("models/cube")
self.cube.reparentTo(self.render)
self.cube.setPos(0, 10, 0)
# Add continuous rotation task
self.taskMgr.add(self.rotateCube, "rotate-cube")
# Add periodic task
self.taskMgr.doMethodLater(
2.0, self.changeColor, "change-color"
)
# Task with completion
self.taskMgr.add(self.moveAndStop, "move-cube")
self.colors = [(1,0,0,1), (0,1,0,1), (0,0,1,1), (1,1,0,1)]
self.colorIndex = 0
self.startTime = globalClock.getFrameTime()
def rotateCube(self, task):
"""Rotate cube continuously."""
dt = globalClock.getDt()
# Rotate around Y axis
currentH = self.cube.getH()
self.cube.setH(currentH + 30 * dt) # 30 degrees per second
return task.cont # Continue running
def changeColor(self, task):
"""Change cube color periodically."""
color = self.colors[self.colorIndex]
self.cube.setColor(*color)
self.colorIndex = (self.colorIndex + 1) % len(self.colors)
# Schedule next color change
task.delayTime = 2.0 # Wait 2 seconds
return task.again # Restart task with delay
def moveAndStop(self, task):
"""Move cube for 5 seconds then stop."""
dt = globalClock.getDt()
elapsed = globalClock.getFrameTime() - self.startTime
if elapsed < 5.0:
# Move cube
currentPos = self.cube.getPos()
self.cube.setPos(currentPos + Vec3(dt, 0, 0))
return task.cont
else:
print("Movement task completed!")
return task.done # Task finished
app = TaskDemo()
app.run()Event-Driven Programming
from direct.showbase.ShowBase import ShowBase
from direct.showbase.DirectObject import DirectObject
from panda3d.core import Vec3
class EventDemo(ShowBase):
def __init__(self):
ShowBase.__init__(self)
# Create game objects
self.player = self.loader.loadModel("models/player")
self.player.reparentTo(self.render)
self.player.setPos(0, 10, 0)
# Create game controller
self.controller = GameController(self.player)
# Setup system event handling
self.accept("escape", self.exitGame)
self.accept("window-close-request", self.exitGame)
# Custom game events
self.accept("player-moved", self.onPlayerMoved)
self.accept("game-over", self.onGameOver)
# Send initial event
messenger.send("game-started")
def onPlayerMoved(self, newPos):
"""Handle player movement event."""
print(f"Player moved to: {newPos}")
# Check win condition
if newPos.getX() > 10:
messenger.send("game-over", ["win"])
def onGameOver(self, result):
"""Handle game over event."""
print(f"Game over: {result}")
if result == "win":
print("Congratulations!")
# Restart after delay
self.taskMgr.doMethodLater(3.0, self.restartGame, "restart")
def restartGame(self, task):
"""Restart the game."""
self.player.setPos(0, 10, 0)
messenger.send("game-started")
return task.done
def exitGame(self):
"""Clean exit."""
print("Exiting game...")
self.userExit()
class GameController(DirectObject):
def __init__(self, player):
DirectObject.__init__(self)
self.player = player
self.speed = 5.0
# Accept input events
self.accept("arrow_left", self.moveLeft)
self.accept("arrow_right", self.moveRight)
self.accept("arrow_up", self.moveUp)
self.accept("arrow_down", self.moveDown)
# Accept game events
self.accept("game-started", self.onGameStart)
# Add update task
self.addTask(self.update, "controller-update")
def moveLeft(self):
"""Move player left."""
pos = self.player.getPos()
newPos = pos + Vec3(-1, 0, 0)
self.player.setPos(newPos)
messenger.send("player-moved", [newPos])
def moveRight(self):
"""Move player right."""
pos = self.player.getPos()
newPos = pos + Vec3(1, 0, 0)
self.player.setPos(newPos)
messenger.send("player-moved", [newPos])
def moveUp(self):
"""Move player forward."""
pos = self.player.getPos()
newPos = pos + Vec3(0, 1, 0)
self.player.setPos(newPos)
messenger.send("player-moved", [newPos])
def moveDown(self):
"""Move player backward."""
pos = self.player.getPos()
newPos = pos + Vec3(0, -1, 0)
self.player.setPos(newPos)
messenger.send("player-moved", [newPos])
def onGameStart(self):
"""Handle game start event."""
print("Game started! Use arrow keys to move.")
def update(self, task):
"""Update controller logic."""
# Could add continuous input handling here
return task.cont
app = EventDemo()
app.run()Advanced Task Patterns
from direct.showbase.ShowBase import ShowBase
from direct.task import Task
class AdvancedTaskDemo(ShowBase):
def __init__(self):
ShowBase.__init__(self)
# Task with state management
self.addTaskWithState()
# Task chains and dependencies
self.addTaskChain()
# Task with cleanup
self.addTaskWithCleanup()
def addTaskWithState(self):
"""Task that maintains state across frames."""
def countingTask(task):
# Initialize state on first run
if not hasattr(task, 'counter'):
task.counter = 0
task.startTime = globalClock.getFrameTime()
task.counter += 1
elapsed = globalClock.getFrameTime() - task.startTime
print(f"Frame {task.counter}, Time: {elapsed:.2f}")
# Stop after 10 seconds
if elapsed > 10.0:
print(f"Counting task completed after {task.counter} frames")
return task.done
return task.cont
self.taskMgr.add(countingTask, "counting-task")
def addTaskChain(self):
"""Chain of dependent tasks."""
def task1(task):
"""First task in chain."""
if not hasattr(task, 'startTime'):
task.startTime = globalClock.getFrameTime()
print("Task 1 started")
elapsed = globalClock.getFrameTime() - task.startTime
if elapsed > 2.0:
print("Task 1 completed, starting Task 2")
# Start next task
self.taskMgr.add(self.task2, "task-2")
return task.done
return task.cont
self.taskMgr.add(task1, "task-1")
def task2(self, task):
"""Second task in chain."""
if not hasattr(task, 'startTime'):
task.startTime = globalClock.getFrameTime()
print("Task 2 started")
elapsed = globalClock.getFrameTime() - task.startTime
if elapsed > 1.5:
print("Task 2 completed, starting Task 3")
# Start final task
self.taskMgr.doMethodLater(0.5, self.task3, "task-3")
return task.done
return task.cont
def task3(self, task):
"""Final task in chain."""
print("Task 3 completed - chain finished!")
return task.done
def addTaskWithCleanup(self):
"""Task with cleanup function."""
def workTask(task):
"""Main work task."""
if not hasattr(task, 'workDone'):
task.workDone = 0
print("Work task started")
task.workDone += 1
print(f"Work progress: {task.workDone}/10")
if task.workDone >= 10:
print("Work task completed")
return task.done
return task.cont
def cleanupTask(task):
"""Cleanup after work completes."""
print("Performing cleanup...")
return task.done
# Add task with cleanup
self.taskMgr.add(
workTask,
"work-task",
uponDeath=cleanupTask
)
app = AdvancedTaskDemo()
app.run()Types
# Task constants and types
Task.cont: int = 0 # Continue task
Task.done: int = 1 # Complete task
Task.again: int = 2 # Restart task
# Global instances
taskMgr: TaskManager # Global task manager
messenger: Messenger # Global messenger
globalClock: ClockObject # Global clock
# Priority constants
SORT_VERY_HIGH = -1000
SORT_HIGH = -100
SORT_DEFAULT = 0
SORT_LOW = 100
SORT_VERY_LOW = 1000Install with Tessl CLI
npx tessl i tessl/pypi-panda3d