Add More Types for mypy

common/realtime.py

@@ -39,7 +39,7 @@ def set_realtime_priority(level: int) -> None:
 
 def set_core_affinity(core: int) -> None:
   if not PC:
-    os.sched_setaffinity(0, [core,])
+    os.sched_setaffinity(0, [core,])   # type: ignore[attr-defined]
 
 
 def config_realtime_process(core: int, priority: int) -> None:

common/spinner.py

@@ -24,7 +24,7 @@ def update(self, spinner_text: str):
       except BrokenPipeError:
         pass
 
-  def update_progress(self, cur: int, total: int):
+  def update_progress(self, cur: float, total: float):
     self.update(str(round(100 * cur / total)))
 
   def close(self):

selfdrive/controls/lib/events.py

@@ -1,5 +1,5 @@
 from enum import IntEnum
-from typing import Dict, Union, Callable
+from typing import Dict, Union, Callable, List, Optional
 
 from cereal import log, car
 import cereal.messaging as messaging
@@ -42,33 +42,33 @@ class ET:
 
 class Events:
   def __init__(self):
-    self.events = []
-    self.static_events = []
+    self.events: List[int] = []
+    self.static_events: List[int] = []
     self.events_prev = dict.fromkeys(EVENTS.keys(), 0)
 
   @property
-  def names(self):
+  def names(self) -> List[int]:
     return self.events
 
-  def __len__(self):
+  def __len__(self) -> int:
     return len(self.events)
 
-  def add(self, event_name, static=False):
+  def add(self, event_name: int, static: bool=False) -> None:
     if static:
       self.static_events.append(event_name)
     self.events.append(event_name)
 
-  def clear(self):
+  def clear(self) -> None:
     self.events_prev = {k: (v + 1 if k in self.events else 0) for k, v in self.events_prev.items()}
     self.events = self.static_events.copy()
 
-  def any(self, event_type):
+  def any(self, event_type: str) -> bool:
     for e in self.events:
       if event_type in EVENTS.get(e, {}).keys():
         return True
     return False
 
-  def create_alerts(self, event_types, callback_args=None):
+  def create_alerts(self, event_types: List[str], callback_args=None):
     if callback_args is None:
       callback_args = []
 
@@ -129,7 +129,7 @@ def __init__(self,
     self.creation_delay = creation_delay
 
     self.alert_type = ""
-    self.event_type = None
+    self.event_type: Optional[str] = None
 
   def __str__(self) -> str:
     return f"{self.alert_text_1}/{self.alert_text_2} {self.priority} {self.visual_alert} {self.audible_alert}"
@@ -139,14 +139,14 @@ def __gt__(self, alert2) -> bool:
 
 
 class NoEntryAlert(Alert):
-  def __init__(self, alert_text_2, visual_alert=VisualAlert.none):
+  def __init__(self, alert_text_2: str, visual_alert: car.CarControl.HUDControl.VisualAlert=VisualAlert.none):
     super().__init__("openpilot Unavailable", alert_text_2, AlertStatus.normal,
                      AlertSize.mid, Priority.LOW, visual_alert,
                      AudibleAlert.refuse, 3.)
 
 
 class SoftDisableAlert(Alert):
-  def __init__(self, alert_text_2):
+  def __init__(self, alert_text_2: str):
     super().__init__("TAKE CONTROL IMMEDIATELY", alert_text_2,
                      AlertStatus.userPrompt, AlertSize.full,
                      Priority.MID, VisualAlert.steerRequired,
@@ -155,13 +155,13 @@ def __init__(self, alert_text_2):
 
 # less harsh version of SoftDisable, where the condition is user-triggered
 class UserSoftDisableAlert(SoftDisableAlert):
-  def __init__(self, alert_text_2):
+  def __init__(self, alert_text_2: str):
     super().__init__(alert_text_2),
     self.alert_text_1 = "openpilot will disengage"
 
 
 class ImmediateDisableAlert(Alert):
-  def __init__(self, alert_text_2):
+  def __init__(self, alert_text_2: str):
     super().__init__("TAKE CONTROL IMMEDIATELY", alert_text_2,
                      AlertStatus.critical, AlertSize.full,
                      Priority.HIGHEST, VisualAlert.steerRequired,

selfdrive/hardware/base.py

@@ -1,11 +1,12 @@
 from abc import abstractmethod, ABC
 from collections import namedtuple
+from typing import Dict
 
 ThermalConfig = namedtuple('ThermalConfig', ['cpu', 'gpu', 'mem', 'bat', 'ambient', 'pmic'])
 
 class HardwareBase(ABC):
   @staticmethod
-  def get_cmdline():
+  def get_cmdline() -> Dict[str, str]:
     with open('/proc/cmdline') as f:
       cmdline = f.read()
     return {kv[0]: kv[1] for kv in [s.split('=') for s in cmdline.split(' ')] if len(kv) == 2}

selfdrive/locationd/calibrationd.py

@@ -8,6 +8,7 @@
 
 import os
 import copy
+from typing import NoReturn
 import numpy as np
 import cereal.messaging as messaging
 from cereal import log
@@ -183,11 +184,11 @@ def get_msg(self):
     msg.liveCalibration.rpyCalibSpread = [float(x) for x in self.calib_spread]
     return msg
 
-  def send_data(self, pm):
+  def send_data(self, pm) -> None:
     pm.send('liveCalibration', self.get_msg())
 
 
-def calibrationd_thread(sm=None, pm=None):
+def calibrationd_thread(sm=None, pm=None) -> NoReturn:
   if sm is None:
     sm = messaging.SubMaster(['cameraOdometry', 'carState'], poll=['cameraOdometry'])
 
@@ -215,7 +216,7 @@ def calibrationd_thread(sm=None, pm=None):
       calibrator.send_data(pm)
 
 
-def main(sm=None, pm=None):
+def main(sm=None, pm=None) -> NoReturn:
   calibrationd_thread(sm, pm)
 
 

selfdrive/manager/build.py

@@ -22,14 +22,15 @@
 PREBUILT = os.path.exists(os.path.join(BASEDIR, 'prebuilt'))
 
 
-def build(spinner, dirty=False):
+def build(spinner: Spinner, dirty: bool = False) -> None:
   env = os.environ.copy()
   env['SCONS_PROGRESS'] = "1"
   nproc = os.cpu_count()
   j_flag = "" if nproc is None else f"-j{nproc - 1}"
 
   for retry in [True, False]:
     scons = subprocess.Popen(["scons", j_flag, "--cache-populate"], cwd=BASEDIR, env=env, stderr=subprocess.PIPE)
+    assert scons.stderr is not None
 
     compile_output = []
 

selfdrive/manager/helpers.py

@@ -5,7 +5,7 @@
 import signal
 
 
-def unblock_stdout():
+def unblock_stdout() -> None:
   # get a non-blocking stdout
   child_pid, child_pty = os.forkpty()
   if child_pid != 0:  # parent

selfdrive/manager/manager.py

@@ -5,6 +5,7 @@
 import subprocess
 import sys
 import traceback
+from typing import List, Tuple, Union
 
 import cereal.messaging as messaging
 import selfdrive.crash as crash
@@ -25,7 +26,7 @@
 sys.path.append(os.path.join(BASEDIR, "pyextra"))
 
 
-def manager_init():
+def manager_init() -> None:
   # update system time from panda
   set_time(cloudlog)
 
@@ -35,7 +36,7 @@ def manager_init():
   params = Params()
   params.clear_all(ParamKeyType.CLEAR_ON_MANAGER_START)
 
-  default_params = [
+  default_params: List[Tuple[str, Union[str, bytes]]] = [
     ("CompletedTrainingVersion", "0"),
     ("HasAcceptedTerms", "0"),
     ("OpenpilotEnabledToggle", "1"),
@@ -56,7 +57,7 @@ def manager_init():
 
   # is this dashcam?
   if os.getenv("PASSIVE") is not None:
-    params.put_bool("Passive", bool(int(os.getenv("PASSIVE"))))
+    params.put_bool("Passive", bool(int(os.getenv("PASSIVE", "0"))))
 
   if params.get("Passive") is None:
     raise Exception("Passive must be set to continue")
@@ -99,12 +100,12 @@ def manager_init():
                    device=HARDWARE.get_device_type())
 
 
-def manager_prepare():
+def manager_prepare() -> None:
   for p in managed_processes.values():
     p.prepare()
 
 
-def manager_cleanup():
+def manager_cleanup() -> None:
   # send signals to kill all procs
   for p in managed_processes.values():
     p.stop(block=False)
@@ -116,7 +117,7 @@ def manager_cleanup():
   cloudlog.info("everything is dead")
 
 
-def manager_thread():
+def manager_thread() -> None:
   cloudlog.bind(daemon="manager")
   cloudlog.info("manager start")
   cloudlog.info({"environ": os.environ})
@@ -128,8 +129,7 @@ def manager_thread():
     ignore += ["manage_athenad", "uploader"]
   if os.getenv("NOBOARD") is not None:
     ignore.append("pandad")
-  if os.getenv("BLOCK") is not None:
-    ignore += os.getenv("BLOCK").split(",")
+  ignore += [x for x in os.getenv("BLOCK", "").split(",") if len(x) > 0]
 
   ensure_running(managed_processes.values(), started=False, not_run=ignore)
 
@@ -176,7 +176,7 @@ def manager_thread():
       break
 
 
-def main():
+def main() -> None:
   prepare_only = os.getenv("PREPAREONLY") is not None
 
   manager_init()