showbase: Annotate methods of Event, Interval, and Job Managers (#1762)

direct/src/interval/IntervalManager.py

@@ -1,11 +1,14 @@
 """Defines the IntervalManager class as well as the global instance of
 this class, ivalMgr."""
 
+from __future__ import annotations
+
 __all__ = ['IntervalManager', 'ivalMgr']
 
 from panda3d.core import EventQueue
-from panda3d.direct import CIntervalManager, Dtool_BorrowThisReference
+from panda3d.direct import CInterval, CIntervalManager, Dtool_BorrowThisReference
 from direct.showbase import EventManager
+from . import Interval
 import fnmatch
 
 class IntervalManager(CIntervalManager):
@@ -15,7 +18,7 @@ class IntervalManager(CIntervalManager):
     # the Python extensions is to add support for Python-based
     # intervals (like MetaIntervals).
 
-    def __init__(self, globalPtr = 0):
+    def __init__(self, globalPtr: bool = False) -> None:
         # Pass globalPtr == 1 to the constructor to trick it into
         # "constructing" a Python wrapper around the global
         # CIntervalManager object.
@@ -28,8 +31,8 @@ class IntervalManager(CIntervalManager):
         self.eventQueue = EventQueue()
         self.MyEventmanager = EventManager.EventManager(self.eventQueue)
         self.setEventQueue(self.eventQueue)
-        self.ivals = []
-        self.removedIvals = {}
+        self.ivals: list[Interval.Interval | CInterval | None] = []
+        self.removedIvals: dict = {}
 
     def addInterval(self, interval):
         index = self.addCInterval(interval, 1)
@@ -86,7 +89,7 @@ class IntervalManager(CIntervalManager):
             ival.pause()
         return len(ivals)
 
-    def step(self):
+    def step(self) -> None:
         # This method should be called once per frame to perform all
         # of the per-frame processing on the active intervals.
         # Call C++ step, then do the Python stuff.
@@ -101,7 +104,7 @@ class IntervalManager(CIntervalManager):
         CIntervalManager.interrupt(self)
         self.__doPythonCallbacks()
 
-    def __doPythonCallbacks(self):
+    def __doPythonCallbacks(self) -> None:
         # This method does all of the required Python post-processing
         # after performing some C++-level action.
         # It is important to call all of the python callbacks on the
@@ -137,4 +140,4 @@ class IntervalManager(CIntervalManager):
         self.ivals[index] = interval
 
 #: The global IntervalManager object.
-ivalMgr = IntervalManager(1)
+ivalMgr = IntervalManager(True)

direct/src/showbase/EventManager.py

@@ -1,21 +1,33 @@
 """Contains the EventManager class.  See :mod:`.EventManagerGlobal` for the
 global eventMgr instance."""
 
+from __future__ import annotations
+
 __all__ = ['EventManager']
 
 
+from typing import Any
+
 from direct.directnotify.DirectNotifyGlobal import directNotify
+from direct.directnotify.Notifier import Notifier
 from direct.task.TaskManagerGlobal import taskMgr
 from direct.showbase.MessengerGlobal import messenger
-from panda3d.core import PStatCollector, EventQueue, EventHandler
-from panda3d.core import ConfigVariableBool
+from panda3d.core import (
+    ConfigVariableBool,
+    Event,
+    EventHandler,
+    EventParameter,
+    EventQueue,
+    PStatCollector,
+    PythonTask,
+)
 
 
 class EventManager:
 
-    notify = None
+    notify: Notifier | None = None
 
-    def __init__(self, eventQueue = None):
+    def __init__(self, eventQueue: EventQueue | None = None) -> None:
         """
         Create a C++ event queue and handler
         """
@@ -24,11 +36,11 @@ class EventManager:
             EventManager.notify = directNotify.newCategory("EventManager")
 
         self.eventQueue = eventQueue
-        self.eventHandler = None
+        self.eventHandler: EventHandler | None = None
 
         self._wantPstats = ConfigVariableBool('pstats-eventmanager', False)
 
-    def doEvents(self):
+    def doEvents(self) -> None:
         """
         Process all the events on the C++ event queue
         """
@@ -38,12 +50,13 @@ class EventManager:
             processFunc = self.processEventPstats
         else:
             processFunc = self.processEvent
+        assert self.eventQueue is not None
         isEmptyFunc = self.eventQueue.isQueueEmpty
         dequeueFunc = self.eventQueue.dequeueEvent
         while not isEmptyFunc():
             processFunc(dequeueFunc())
 
-    def eventLoopTask(self, task):
+    def eventLoopTask(self, task: PythonTask) -> int:
         """
         Process all the events on the C++ event queue
         """
@@ -51,7 +64,7 @@ class EventManager:
         messenger.send("event-loop-done")
         return task.cont
 
-    def parseEventParameter(self, eventParameter):
+    def parseEventParameter(self, eventParameter: EventParameter) -> Any:
         """
         Extract the actual data from the eventParameter
         """
@@ -72,7 +85,7 @@ class EventManager:
             # which will be downcast to that type.
             return eventParameter.getPtr()
 
-    def processEvent(self, event):
+    def processEvent(self, event: Event) -> None:
         """
         Process a C++ event
         Duplicate any changes in processEventPstats
@@ -89,6 +102,7 @@ class EventManager:
                 paramList.append(eventParameterData)
 
             # Do not print the new frame debug, it is too noisy!
+            assert EventManager.notify is not None
             if EventManager.notify.getDebug() and eventName != 'NewFrame':
                 EventManager.notify.debug('received C++ event named: ' + eventName +
                                           ' parameters: ' + repr(paramList))
@@ -106,9 +120,10 @@ class EventManager:
 
         else:
             # An unnamed event from C++ is probably a bad thing
+            assert EventManager.notify is not None
             EventManager.notify.warning('unnamed event in processEvent')
 
-    def processEventPstats(self, event):
+    def processEventPstats(self, event: Event) -> None:
         """
         Process a C++ event with pstats tracking
         Duplicate any changes in processEvent
@@ -125,6 +140,7 @@ class EventManager:
                 paramList.append(eventParameterData)
 
             # Do not print the new frame debug, it is too noisy!
+            assert EventManager.notify is not None
             if EventManager.notify.getDebug() and eventName != 'NewFrame':
                 EventManager.notify.debug('received C++ event named: ' + eventName +
                                           ' parameters: ' + repr(paramList))
@@ -156,9 +172,10 @@ class EventManager:
 
         else:
             # An unnamed event from C++ is probably a bad thing
+            assert EventManager.notify is not None
             EventManager.notify.warning('unnamed event in processEvent')
 
-    def restart(self):
+    def restart(self) -> None:
         if self.eventQueue is None:
             self.eventQueue = EventQueue.getGlobalEventQueue()
 
@@ -173,7 +190,7 @@ class EventManager:
 
         taskMgr.add(self.eventLoopTask, 'eventManager')
 
-    def shutdown(self):
+    def shutdown(self) -> None:
         taskMgr.remove('eventManager')
 
         # Flush the event queue.  We do this after removing the task

direct/src/showbase/JobManager.py

@@ -1,3 +1,7 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+
 from panda3d.core import ConfigVariableBool, ConfigVariableDouble, ClockObject
 from direct.directnotify.DirectNotifyGlobal import directNotify
 from direct.task.TaskManagerGlobal import taskMgr
@@ -17,28 +21,28 @@ class JobManager:
     # there's one task for the JobManager, all jobs run in this task
     TaskName = 'jobManager'
 
-    def __init__(self, timeslice=None):
+    def __init__(self, timeslice: float | None = None) -> None:
         # how long do we run per frame
         self._timeslice = timeslice
         # store the jobs in these structures to allow fast lookup by various keys
         # priority -> jobId -> job
-        self._pri2jobId2job = {}
+        self._pri2jobId2job: dict[int, dict[int, Job]] = {}
         # priority -> chronological list of jobIds
-        self._pri2jobIds = {}
+        self._pri2jobIds: dict[int, list[int]] = {}
         # jobId -> priority
-        self._jobId2pri = {}
+        self._jobId2pri: dict[int, int] = {}
         # how many timeslices to give each job; this is used to efficiently implement
         # the relative job priorities
-        self._jobId2timeslices = {}
+        self._jobId2timeslices: dict[int, int] = {}
         # how much time did the job use beyond the allotted timeslice, used to balance
         # out CPU usage
-        self._jobId2overflowTime = {}
-        self._useOverflowTime = None
+        self._jobId2overflowTime: dict[int, float] = {}
+        self._useOverflowTime: bool | None = None
         # this is a generator that we use to give high-priority jobs more timeslices,
         # it yields jobIds in a sequence that includes high-priority jobIds more often
         # than low-priority
-        self._jobIdGenerator = None
-        self._highestPriority = Job.Priorities.Normal
+        self._jobIdGenerator: Iterator[int] | None = None
+        self._highestPriority: int = Job.Priorities.Normal  # type: ignore[attr-defined]
 
     def destroy(self):
         taskMgr.remove(JobManager.TaskName)