Cleaning up code.

Signed-off-by: Mike Stitt <[email protected]>

Author: MikeStitt

subprojects/robotpy-wpilib/wpilib/iterativerobotpy.py

@@ -37,74 +37,74 @@ def __init__(self, period: float):
         self._teleopPeriodicHasRun: bool = False
         self._testPeriodicHasRun: bool = False
 
-    def robotInit(self):
+    def robotInit(self) -> None:
         pass
 
-    def driverStationConnected(self):
+    def driverStationConnected(self) -> None:
         pass
 
-    def simulationInit(self):
+    def simulationInit(self) -> None:
         pass
 
-    def disabledInit(self):
+    def disabledInit(self) -> None:
         pass
 
-    def autonomousInit(self):
+    def autonomousInit(self) -> None:
         pass
 
-    def teleopInit(self):
+    def teleopInit(self) -> None:
         pass
 
-    def testInit(self):
+    def testInit(self) -> None:
         pass
 
-    def robotPeriodic(self):
+    def robotPeriodic(self) -> None:
         if not self._robotPeriodicHasRun:
             print(f"Default RobotPeriodic() method...Override me!")
             self._robotPeriodicHasRun = True
 
-    def simulationPeriodic(self):
+    def simulationPeriodic(self) -> None:
         if not self._simulationPeriodicHasRun:
             print(f"Default simulationPeriodic() method...Override me!")
             self._simulationPeriodicHasRun = True
 
-    def disabledPeriodic(self):
+    def disabledPeriodic(self) -> None:
         if not self._disabledPeriodicHasRun:
             print(f"Default disabledPeriodic() method...Override me!")
             self._disabledPeriodicHasRun = True
 
-    def autonomousPeriodic(self):
+    def autonomousPeriodic(self) -> None:
         if not self._autonomousPeriodicHasRun:
             print(f"Default autonomousPeriodic() method...Override me!")
             self._autonomousPeriodicHasRun = True
 
-    def teleopPeriodic(self):
+    def teleopPeriodic(self)->None:
         if not self._teleopPeriodicHasRun:
             print(f"Default teleopPeriodic() method...Override me!")
             self._teleopPeriodicHasRun = True
 
-    def testPeriodic(self):
+    def testPeriodic(self)->None:
         if not self._testPeriodicHasRun:
             print(f"Default testPeriodic() method...Override me!")
             self._teleopPeriodicHasRun = True
 
-    def disabledExit(self):
+    def disabledExit(self)->None:
         pass
 
-    def autonomousExit(self):
+    def autonomousExit(self)->None:
         pass
 
-    def teleopExit(self):
+    def teleopExit(self)->None:
         pass
 
-    def testExit(self):
+    def testExit(self)->None:
         pass
 
     # todo @Deprecated(forRemoval=true, since="2025")
-    def setNetworkTablesFlushEnabled(self, enabled: bool):
+    def setNetworkTablesFlushEnabled(self, enabled: bool)->None:
         self._ntFlushEnabled = enabled
 
-    def enableLiveWindowInTest(self, testLW: bool):
+    def enableLiveWindowInTest(self, testLW: bool)->None:
         if self.isTestEnabled():
             raise RuntimeError("Can't configure test mode while in test mode!")
         if not self._reportedLw and testLW:
@@ -118,7 +118,7 @@ def isLiveWindowEnabledInTest(self) -> bool:
     def getPeriod(self) -> float:
         return self._periodS
 
-    def loopFunc(self):
+    def loopFunc(self)->None:
         DriverStation.refreshData()
         self.watchdog.reset()
 
@@ -248,10 +248,10 @@ def loopFunc(self):
         if self.watchdog.isExpired():
             self.printWatchdogEpochs()
 
-    def printLoopOverrunMessage(self):
+    def printLoopOverrunMessage(self)->None:
         reportWarning(
             f"Loop time of {self.watchdog.getTimeout()}s overrun", False
         )
 
-    def printWatchdogEpochs(self):
+    def printWatchdogEpochs(self)->None:
         self.watchdog.printEpochs()

subprojects/robotpy-wpilib/wpilib/timedrobotpy.py

@@ -1,3 +1,4 @@
+from typing import Any, Callable, Iterable
 from heapq import heappush, heappop
 from hal import report, initializeNotifier, setNotifierName, observeUserProgramStarting, updateNotifierAlarm, \
     waitForNotifierAlarm, stopNotifier, tResourceType, tInstances
@@ -10,20 +11,20 @@
 _kFramework_Timed = tInstances.kFramework_Timed
 
 class _Callback:
-    def __init__(self, func, periodUs: int, expirationUs: int):
+    def __init__(self, func: Callable[[],None], periodUs: int, expirationUs: int):
         self.func = func
         self._periodUs = periodUs
         self.expirationUs = expirationUs
 
     @classmethod
     def makeCallBack(cls,
-                     func,
+                     func: Callable[[],None],
                      startTimeUs: int,
                      periodUs: int,
-                     offsetUs: int):
+                     offsetUs: int) -> "_Callback":
 
         callback = _Callback(
-            func,
+            func=func,
             periodUs=periodUs,
             expirationUs=startTimeUs
         )
@@ -42,42 +43,42 @@ def calcFutureExpirationUs(self, currentTimeUs: int) -> int:
         return self.expirationUs + self._periodUs + \
             ((currentTimeUs - self.expirationUs) // self._periodUs) * self._periodUs
 
-    def setNextStartTimeUs(self, currentTimeUs: int):
+    def setNextStartTimeUs(self, currentTimeUs: int) -> None:
         self.expirationUs = self.calcFutureExpirationUs(currentTimeUs)
 
-    def __lt__(self, other):
+    def __lt__(self, other) -> bool:
         return self.expirationUs < other.expirationUs
 
-    def __bool__(self):
+    def __bool__(self) -> bool:
         return True
 
 
 class _OrderedList:
     def __init__(self):
-        self._data = []
+        self._data: list[Any] = []
 
-    def add(self, item):
+    def add(self, item: Any) -> None:
         heappush(self._data, item)
 
-    def pop(self):
+    def pop(self) -> Any:
         return heappop(self._data)
 
-    def peek(self):
+    def peek(self) -> Any|None:
         if self._data:
             return self._data[0]
         else:
             return None
 
-    def __len__(self):
+    def __len__(self) -> int:
         return len(self._data)
 
-    def __iter__(self):
+    def __iter__(self) -> Iterable[Any]:
         return iter(sorted(self._data))
 
-    def __contains__(self, item):
+    def __contains__(self, item) -> bool:
         return item in self._data
 
-    def __str__(self):
+    def __str__(self) -> str:
         return str(sorted(self._data))
 
 
@@ -93,13 +94,11 @@ def __init__(self, period: float = 0.020):
 
         self._notifier, status = initializeNotifier()
         if status != 0:
-            message = f"initializeNotifier() returned {status} {self._notifier}"
-            #raise RuntimeError(message) # todo
+            raise RuntimeError(f"initializeNotifier() returned {self._notifier}, {status}")
 
         status = setNotifierName(self._notifier, "TimedRobot")
         if status != 0:
-            message = f"setNotifierName() returned {status}"
-            #raise RuntimeError(message) # todo
+            raise RuntimeError(f"setNotifierName() returned {status}")
 
         report(_kResourceType_Framework, _kFramework_Timed)
 
@@ -123,13 +122,11 @@ def startCompetition(self) -> None:
 
             status = updateNotifierAlarm(self._notifier, callback.expirationUs)
             if status != 0:
-                message = f"updateNotifierAlarm() returned {status}"
-                #raise RuntimeError(message) # todo
+                raise RuntimeError(f"updateNotifierAlarm() returned {status}")
 
             currentTimeUs, status = waitForNotifierAlarm(self._notifier)
             if status != 0:
-                message = f"waitForNotifierAlarm() returned currentTimeUs={currentTimeUs} status={status}"
-                #raise RuntimeError(message) # todo
+                raise RuntimeError(f"waitForNotifierAlarm() returned currentTimeUs={currentTimeUs} status={status}")
 
             if currentTimeUs == 0:
                 # when HAL_StopNotifier(self.notifier) is called the above waitForNotifierAlarm
@@ -145,33 +142,22 @@ def startCompetition(self) -> None:
                 callback = self._callbacks.pop()
                 self._runCallbackAndReschedule(callback, currentTimeUs)
 
-    def _runCallbackAndReschedule(self, callback, currentTimeUs:int):
+    def _runCallbackAndReschedule(self, callback: Callable[[],None], currentTimeUs: int) -> None:
         callback.func()
         callback.setNextStartTimeUs(currentTimeUs)
         self._callbacks.add(callback)
 
-    def endCompetition(self):
+    def endCompetition(self) -> None:
         stopNotifier(self._notifier)
 
-    """
-    todo this doesn't really translate to python (is it really needed?):    
-
-    TimedRobot::~TimedRobot() {
-      if (m_notifier != HAL_kInvalidHandle) {
-        int32_t status = 0;
-        HAL_StopNotifier(m_notifier, &status);
-        FRC_ReportError(status, "StopNotifier");
-      }
-    }
-    """
-
-    def getLoopStartTime(self):
-        return self.loopStartTimeUs/1e6  # todo units are seconds
+    def getLoopStartTime(self) -> float:
+        return self.loopStartTimeUs/1e6  # units are seconds
 
     def addPeriodic(self,
-                    callback,  # todo typehint
-                    period: float,  # todo units seconds
-                    offset: float = 0.0):  # todo units seconds
+                    callback: Callable[[],None],
+                    period: float, # units are seconds
+                    offset: float = 0.0 # units are seconds
+                    ) -> None:
         self._callbacks.add(
             _Callback.makeCallBack(
                 callback,