Run mypy commit hook (#1591)

* run mypy commit hook

* fix mypy errors
old-commit-hash: 3d08dcc3b2

.gitignore

@@ -61,3 +61,4 @@ cppcheck_report.txt
 htmlcov
 pandaextra
 
+.mypy_cache/

.pre-commit-config.yaml

@@ -6,6 +6,11 @@ repos:
     -   id: check-json
     -   id: check-xml
     -   id: check-yaml
+-   repo: https://github.com/pre-commit/mirrors-mypy
+    rev: master
+    hooks:
+    -   id: mypy
+        exclude: '^(pyextra)|(external)|(cereal)|(rednose)|(panda)|(laika)|(laika_repo)|(rednose_repo)/'
 -   repo: https://gitlab.com/PyCQA/flake8
     rev: master
     hooks:

Pipfile

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:c4ea33acc8b1f639f0719f436989886263a67271decbcf29ab8907208161da03
-size 2043
+oid sha256:7d81ff54962f8734b5c22f96167c7c4b74dfc09234e401c7d7e9b16db4a6a48f
+size 2054

Pipfile.lock

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:34b9162abbc667c0b14f418f43668422491613c32bfe8a7d8da86b3fc256350f
-size 165941
+oid sha256:93a390d4adf714de0f790b782b83b82301009f2ec74599b675bceb4b79d10fd7
+size 170223

common/params.py

@@ -22,10 +22,7 @@ file in place without messing with <params_dir>/d.
 """
 import time
 import os
-import string
-import binascii
 import errno
-import sys
 import shutil
 import fcntl
 import tempfile
@@ -401,22 +398,3 @@ def put_nonblocking(key, val):
   t = threading.Thread(target=f, args=(key, val))
   t.start()
   return t
-
-
-if __name__ == "__main__":
-  params = Params()
-  if len(sys.argv) > 2:
-    params.put(sys.argv[1], sys.argv[2])
-  else:
-    for k in keys:
-      pp = params.get(k)
-      if pp is None:
-        print("%s is None" % k)
-      elif all(chr(c) in string.printable for c in pp):
-        print("%s = %s" % (k, pp))
-      else:
-        print("%s = %s" % (k, binascii.hexlify(pp)))
-
-  # Test multiprocess:
-  # seq 0 100000 | xargs -P20 -I{} python common/params.py DongleId {} && sleep 0.05
-  # while python common/params.py DongleId; do sleep 0.05; done

common/spinner.py

@@ -40,7 +40,7 @@ class Spinner():
     self.close()
 
 
-class FakeSpinner():
+class FakeSpinner(Spinner):
   def __init__(self):
     pass
 

common/text_window.py

@@ -43,7 +43,7 @@ class TextWindow():
     self.close()
 
 
-class FakeTextWindow():
+class FakeTextWindow(TextWindow):
   def __init__(self, s):
     pass
 

mypy.ini

@@ -0,0 +1,4 @@
+[mypy]
+python_version = 3.8
+ignore_missing_imports = True
+

selfdrive/athena/athenad.py

@@ -1,29 +1,31 @@
 #!/usr/bin/env python3
-import json
-import os
+import base64
 import hashlib
 import io
+import json
+import os
+import queue
 import random
 import select
 import socket
-import time
 import threading
-import base64
-import requests
-import queue
+import time
 from collections import namedtuple
 from functools import partial
+from typing import Any
+
+import requests
 from jsonrpc import JSONRPCResponseManager, dispatcher
-from websocket import create_connection, WebSocketTimeoutException, ABNF
-from selfdrive.loggerd.config import ROOT
+from websocket import ABNF, WebSocketTimeoutException, create_connection
 
 import cereal.messaging as messaging
+from cereal.services import service_list
 from common import android
-from common.basedir import PERSIST
 from common.api import Api
+from common.basedir import PERSIST
 from common.params import Params
 from common.realtime import sec_since_boot
-from cereal.services import service_list
+from selfdrive.loggerd.config import ROOT
 from selfdrive.swaglog import cloudlog
 
 ATHENA_HOST = os.getenv('ATHENA_HOST', 'wss://athena.comma.ai')
@@ -31,10 +33,10 @@ HANDLER_THREADS = os.getenv('HANDLER_THREADS', 4)
 LOCAL_PORT_WHITELIST = set([8022])
 
 dispatcher["echo"] = lambda s: s
-payload_queue = queue.Queue()
-response_queue = queue.Queue()
-upload_queue = queue.Queue()
-cancelled_uploads = set()
+payload_queue: Any = queue.Queue()
+response_queue: Any = queue.Queue()
+upload_queue: Any = queue.Queue()
+cancelled_uploads: Any = set()
 UploadItem = namedtuple('UploadItem', ['path', 'url', 'headers', 'created_at', 'id'])
 
 def handle_long_poll(ws):

selfdrive/car/fw_versions.py

@@ -1,15 +1,17 @@
 #!/usr/bin/env python3
-import traceback
 import struct
+import traceback
+from typing import Any
+
 from tqdm import tqdm
 
+import panda.python.uds as uds
+from cereal import car
+from selfdrive.car.fingerprints import FW_VERSIONS, get_attr_from_cars
 from selfdrive.car.isotp_parallel_query import IsoTpParallelQuery
-from selfdrive.swaglog import cloudlog
-from selfdrive.car.fingerprints import get_attr_from_cars, FW_VERSIONS
 from selfdrive.car.toyota.values import CAR as TOYOTA
-import panda.python.uds as uds
+from selfdrive.swaglog import cloudlog
 
-from cereal import car
 Ecu = car.CarParams.Ecu
 
 
@@ -213,7 +215,7 @@ if __name__ == "__main__":
   logcan = messaging.sub_sock('can')
   sendcan = messaging.pub_sock('sendcan')
 
-  extra = None
+  extra: Any = None
   if args.scan:
     extra = {}
     # Honda

selfdrive/crash.py

@@ -9,7 +9,7 @@ from selfdrive.swaglog import cloudlog
 from common.android import ANDROID
 
 if os.getenv("NOLOG") or os.getenv("NOCRASH") or not ANDROID:
-  def capture_exception(*exc_info):
+  def capture_exception(*args, **kwargs):
     pass
   def bind_user(**kwargs):
     pass

selfdrive/debug/check_freq.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+# type: ignore
+
 import argparse
 import numpy as np
 from collections import defaultdict, deque

selfdrive/debug/check_lag.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+# type: ignore
+
 import cereal.messaging as messaging
 from cereal.services import service_list
 

selfdrive/debug/cpu_usage_stat.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+# type: ignore
 import psutil
 import time
 import os

selfdrive/debug/internal/get_fingerprint_from_route.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+# type: ignore
 
 import sys
 from tools.lib.logreader import MultiLogIterator

selfdrive/debug/internal/measure_steering_accuracy.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+# type: ignore
+
 import os
 import argparse
 import signal

selfdrive/debug/internal/measure_torque_time_to_max.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+# type: ignore
+
 import os
 import argparse
 import struct

selfdrive/debug/mpc/test_mpc_wobble.py

@@ -1,4 +1,5 @@
 #! /usr/bin/env python
+# type: ignore
 import matplotlib.pyplot as plt
 from selfdrive.controls.lib.lateral_mpc import libmpc_py
 from selfdrive.controls.lib.drive_helpers import MPC_COST_LAT

selfdrive/debug/mpc/tune_lateral.py

@@ -1,4 +1,6 @@
 #! /usr/bin/env python
+# type: ignore
+
 import numpy as np
 from collections import OrderedDict
 import matplotlib.pyplot as plt

selfdrive/debug/mpc/tune_longitudinal.py

@@ -1,4 +1,5 @@
 #! /usr/bin/env python
+# type: ignore
 import numpy as np
 import matplotlib.pyplot as plt
 from selfdrive.controls.lib.longitudinal_mpc import libmpc_py

selfdrive/debug/test_fw_query_on_routes.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+# type: ignore
+
 import argparse
 import os
 import traceback

selfdrive/locationd/models/car_kf.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
-import sys
 import math
+import sys
+from typing import Any, Dict
 
 import numpy as np
 import sympy as sp
@@ -58,7 +59,7 @@ class CarKalman(KalmanFilter):
   ])
   P_initial = Q.copy()
 
-  obs_noise = {
+  obs_noise : Dict[int, Any] = {
     ObservationKind.STEER_ANGLE: np.atleast_2d(math.radians(0.01)**2),
     ObservationKind.ANGLE_OFFSET_FAST: np.atleast_2d(math.radians(10.0)**2),
     ObservationKind.STEER_RATIO: np.atleast_2d(5.0**2),

selfdrive/locationd/models/gnss_kf.py

@@ -1,11 +1,12 @@
 #!/usr/bin/env python3
 import sys
+from typing import List
 
 import numpy as np
 import sympy as sp
 
-from selfdrive.locationd.models.constants import ObservationKind
 from rednose.helpers.ekf_sym import EKF_sym, gen_code
+from selfdrive.locationd.models.constants import ObservationKind
 from selfdrive.locationd.models.loc_kf import parse_pr, parse_prr
 
 
@@ -39,7 +40,7 @@ class GNSSKalman():
                (.1)**2, (0)**2, (0.005)**2,
                .1**2, (.01)**2])
 
-  maha_test_kinds = []  # ObservationKind.PSEUDORANGE_RATE, ObservationKind.PSEUDORANGE, ObservationKind.PSEUDORANGE_GLONASS]
+  maha_test_kinds: List[int] = []  # ObservationKind.PSEUDORANGE_RATE, ObservationKind.PSEUDORANGE, ObservationKind.PSEUDORANGE_GLONASS]
 
   @staticmethod
   def generate_code(generated_dir):

selfdrive/locationd/test/ci_test.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+# type: ignore
 import subprocess
 import os
 import sys

selfdrive/locationd/test/ubloxd.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+# type: ignore
 
 import os
 import serial

selfdrive/locationd/test/ubloxd_easy.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+# type: ignore
 
 import os
 from selfdrive.locationd.test import ublox

selfdrive/locationd/test/ubloxd_py_test.py

@@ -1,3 +1,5 @@
+# type: ignore
+
 import sys
 import os
 

selfdrive/manager.py

@@ -9,8 +9,10 @@ import shutil
 import subprocess
 import datetime
 import textwrap
+from typing import Dict, List
 from selfdrive.swaglog import cloudlog, add_logentries_handler
 
+
 from common.basedir import BASEDIR, PARAMS
 from common.android import ANDROID
 WEBCAM = os.getenv("WEBCAM") is not None
@@ -99,7 +101,7 @@ if not prebuilt:
     # Read progress from stderr and update spinner
     while scons.poll() is None:
       try:
-        line = scons.stderr.readline()
+        line = scons.stderr.readline()  # type: ignore
         if line is None:
           continue
         line = line.rstrip()
@@ -117,7 +119,7 @@ if not prebuilt:
 
     if scons.returncode != 0:
       # Read remaining output
-      r = scons.stderr.read().split(b'\n')
+      r = scons.stderr.read().split(b'\n')   # type: ignore
       compile_output += r
 
       if retry:
@@ -194,7 +196,7 @@ daemon_processes = {
   "manage_athenad": ("selfdrive.athena.manage_athenad", "AthenadPid"),
 }
 
-running = {}
+running: Dict[str, Process] = {}
 def get_running():
   return running
 
@@ -202,7 +204,7 @@ def get_running():
 unkillable_processes = ['camerad']
 
 # processes to end with SIGINT instead of SIGTERM
-interrupt_processes = []
+interrupt_processes: List[str] = []
 
 # processes to end with SIGKILL instead of SIGTERM
 kill_processes = ['sensord', 'paramsd']

selfdrive/test/longitudinal_maneuvers/plant_ui.py

@@ -1,119 +0,0 @@
-#!/usr/bin/env python3
-import pygame # pylint: disable=import-error
-from selfdrive.test.longitudinal_maneuvers.plant import Plant
-from selfdrive.car.honda.values import CruiseButtons
-import numpy as np
-import cereal.messaging as messaging
-import math
-
-CAR_WIDTH = 2.0
-CAR_LENGTH = 4.5
-
-METER = 8
-
-def rot_center(image, angle):
-  """rotate an image while keeping its center and size"""
-  orig_rect = image.get_rect()
-  rot_image = pygame.transform.rotate(image, angle)
-  rot_rect = orig_rect.copy()
-  rot_rect.center = rot_image.get_rect().center
-  rot_image = rot_image.subsurface(rot_rect).copy()
-  return rot_image
-
-def car_w_color(c):
-  car = pygame.Surface((METER*CAR_LENGTH, METER*CAR_LENGTH))  # pylint: disable=too-many-function-args
-  car.set_alpha(0)
-  car.fill((10,10,10))
-  car.set_alpha(128)
-  pygame.draw.rect(car, c, (METER*1.25, 0, METER*CAR_WIDTH, METER*CAR_LENGTH), 1)
-  return car
-
-if __name__ == "__main__":
-  pygame.init()
-  display = pygame.display.set_mode((1000, 1000))
-  pygame.display.set_caption('Plant UI')
-
-  car = car_w_color((255,0,255))
-  leadcar = car_w_color((255,0,0))
-
-  carx, cary, heading = 10.0, 50.0, 0.0
-
-  plant = Plant(100, distance_lead = 40.0)
-
-  control_offset = 2.0
-  control_pts = list(zip(np.arange(0, 100.0, 10.0), [50.0 + control_offset]*10))
-
-  def pt_to_car(pt):
-    x,y = pt
-    x -= carx
-    y -= cary
-    rx = x * math.cos(-heading) + y * -math.sin(-heading)
-    ry = x * math.sin(-heading) + y * math.cos(-heading)
-    return rx, ry
-
-  def pt_from_car(pt):
-    x,y = pt
-    rx = x * math.cos(heading) + y * -math.sin(heading)
-    ry = x * math.sin(heading) + y * math.cos(heading)
-    rx += carx
-    ry += cary
-    return rx, ry
-
-  while 1:
-    if plant.rk.frame%100 >= 20 and plant.rk.frame%100 <= 25:
-      cruise_buttons = CruiseButtons.RES_ACCEL
-    else:
-      cruise_buttons = 0
-
-    md = messaging.new_message('model')
-    md.model.frameId = 0
-    for x in [md.model.path, md.model.leftLane, md.model.rightLane]:
-      x.points = [0.0]*50
-      x.prob = 0.0
-      x.std = 1.0
-
-    car_pts = [pt_to_car(pt) for pt in control_pts]
-
-    print(car_pts)
-
-    car_poly = np.polyfit([x[0] for x in car_pts], [x[1] for x in car_pts], 3)
-    md.model.path.points = np.polyval(car_poly, np.arange(0, 50)).tolist()
-    md.model.path.prob = 1.0
-    Plant.model.send(md.to_bytes())
-
-    plant.step(cruise_buttons = cruise_buttons, v_lead = 2.0, publish_model = False)
-
-    display.fill((10,10,10))
-
-    carx += plant.speed * plant.ts * math.cos(heading)
-    cary += plant.speed * plant.ts * math.sin(heading)
-
-    # positive steering angle = steering right
-    print(plant.angle_steer)
-    heading += plant.angle_steer * plant.ts
-    print(heading)
-
-    # draw my car
-    display.blit(pygame.transform.rotate(car, 90-math.degrees(heading)), (carx*METER, cary*METER))
-
-    # draw control pts
-    for x,y in control_pts:
-      pygame.draw.circle(display, (255,255,0), (int(x * METER),int(y * METER)), 2)
-
-    # draw path
-    path_pts = zip(np.arange(0, 50), md.model.path.points)
-
-    for x,y in path_pts:
-      x,y = pt_from_car((x,y))
-      pygame.draw.circle(display, (0,255,0), (int(x * METER),int(y * METER)), 1)
-
-    """
-    # draw lead car
-    dl = (plant.distance_lead - plant.distance) + 4.5
-    lx = carx + dl * math.cos(heading)
-    ly = cary + dl * math.sin(heading)
-
-    display.blit(pygame.transform.rotate(leadcar, 90-math.degrees(heading)), (lx*METER, ly*METER))
-    """
-
-    pygame.display.flip()

selfdrive/test/process_replay/compare_logs.py

@@ -8,7 +8,7 @@ import dictdiffer
 if "CI" in os.environ:
   tqdm = lambda x: x
 else:
-  from tqdm import tqdm
+  from tqdm import tqdm  # type: ignore
 
 from tools.lib.logreader import LogReader
 

selfdrive/test/process_replay/flycheck_compare_logs.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+import bz2
+import os
+import sys
+import numbers
+
+import dictdiffer
+if "CI" in os.environ:
+  tqdm = lambda x: x
+else:
+  from tqdm import tqdm  # type: ignore
+
+from tools.lib.logreader import LogReader
+
+def save_log(dest, log_msgs):
+  dat = b""
+  for msg in tqdm(log_msgs):
+    dat += msg.as_builder().to_bytes()
+  dat = bz2.compress(dat)
+
+  with open(dest, "wb") as f:
+   f.write(dat)
+
+def remove_ignored_fields(msg, ignore):
+  msg = msg.as_builder()
+  for key in ignore:
+    attr = msg
+    keys = key.split(".")
+    if msg.which() not in key and len(keys) > 1:
+      continue
+
+    for k in keys[:-1]:
+      try:
+        attr = getattr(msg, k)
+      except:
+        break
+    else:
+      v = getattr(attr, keys[-1])
+      if isinstance(v, bool):
+        val = False
+      elif isinstance(v, numbers.Number):
+        val = 0
+      else:
+        raise NotImplementedError
+      setattr(attr, keys[-1], val)
+  return msg.as_reader()
+
+def compare_logs(log1, log2, ignore_fields=[], ignore_msgs=[]):
+  filter_msgs = lambda m: m.which() not in ignore_msgs
+  log1, log2 = [list(filter(filter_msgs, log)) for log in (log1, log2)]
+  assert len(log1) == len(log2), "logs are not same length: " + str(len(log1)) + " VS " + str(len(log2))
+
+  diff = []
+  for msg1, msg2 in tqdm(zip(log1, log2)):
+    if msg1.which() != msg2.which():
+      print(msg1, msg2)
+      raise Exception("msgs not aligned between logs")
+
+    msg1_bytes = remove_ignored_fields(msg1, ignore_fields).as_builder().to_bytes()
+    msg2_bytes = remove_ignored_fields(msg2, ignore_fields).as_builder().to_bytes()
+
+    if msg1_bytes != msg2_bytes:
+      msg1_dict = msg1.to_dict(verbose=True)
+      msg2_dict = msg2.to_dict(verbose=True)
+      dd = dictdiffer.diff(msg1_dict, msg2_dict, ignore=ignore_fields, tolerance=0)
+      diff.extend(dd)
+  return diff
+
+if __name__ == "__main__":
+  log1 = list(LogReader(sys.argv[1]))
+  log2 = list(LogReader(sys.argv[2]))
+  print(compare_logs(log1, log2, sys.argv[3:]))

selfdrive/test/process_replay/process_replay.py

@@ -7,7 +7,7 @@ import importlib
 if "CI" in os.environ:
   tqdm = lambda x: x
 else:
-  from tqdm import tqdm
+  from tqdm import tqdm   # type: ignore
 
 from cereal import car, log
 from selfdrive.car.car_helpers import get_car

selfdrive/test/process_replay/test_processes.py

@@ -2,13 +2,14 @@
 import argparse
 import os
 import sys
+from typing import Any
 
 from selfdrive.car.car_helpers import interface_names
-from selfdrive.test.process_replay.process_replay import replay_process, CONFIGS
 from selfdrive.test.process_replay.compare_logs import compare_logs
+from selfdrive.test.process_replay.process_replay import (CONFIGS,
+                                                          replay_process)
 from tools.lib.logreader import LogReader
 
-
 INJECT_MODEL = 0
 
 segments = [
@@ -134,7 +135,7 @@ if __name__ == "__main__":
     untested = (set(interface_names) - set(excluded_interfaces)) - tested_cars
     assert len(untested) == 0, "Cars missing routes: %s" % (str(untested))
 
-  results = {}
+  results: Any = {}
   for car_brand, segment in segments:
     if (cars_whitelisted and car_brand.upper() not in args.whitelist_cars) or \
         (not cars_whitelisted and car_brand.upper() in args.blacklist_cars):

selfdrive/test/test_car_models.py

@@ -1,27 +1,28 @@
 #!/usr/bin/env python3
-import time
 import os
-import sys
 import signal
 import subprocess
+import sys
+import time
+from typing import List, cast
+
 import requests
-from cereal import car
 
-import selfdrive.manager as manager
 import cereal.messaging as messaging
-from common.params import Params
+import selfdrive.manager as manager
+from cereal import car
 from common.basedir import BASEDIR
+from common.params import Params
+from selfdrive.car.chrysler.values import CAR as CHRYSLER
 from selfdrive.car.fingerprints import all_known_cars
-from selfdrive.car.honda.values import CAR as HONDA
-from selfdrive.car.toyota.values import CAR as TOYOTA
-from selfdrive.car.gm.values import CAR as GM
 from selfdrive.car.ford.values import CAR as FORD
+from selfdrive.car.gm.values import CAR as GM
+from selfdrive.car.honda.values import CAR as HONDA
 from selfdrive.car.hyundai.values import CAR as HYUNDAI
-from selfdrive.car.chrysler.values import CAR as CHRYSLER
+from selfdrive.car.nissan.values import CAR as NISSAN
 from selfdrive.car.subaru.values import CAR as SUBARU
+from selfdrive.car.toyota.values import CAR as TOYOTA
 from selfdrive.car.volkswagen.values import CAR as VOLKSWAGEN
-from selfdrive.car.nissan.values import CAR as NISSAN
-
 
 os.environ['NOCRASH'] = '1'
 
@@ -363,7 +364,7 @@ routes = {
   },
 }
 
-passive_routes = [
+passive_routes: List[str] = [
 ]
 
 forced_dashcam_routes = [
@@ -436,7 +437,7 @@ if __name__ == "__main__":
 
     os.environ['SKIP_FW_QUERY'] = "1"
     if checks.get('fingerprintSource', None) == 'fixed':
-      os.environ['FINGERPRINT'] = checks['carFingerprint']
+      os.environ['FINGERPRINT'] = cast(str, checks['carFingerprint'])
     else:
       os.environ['FINGERPRINT'] = ""
 

tools/lib/auth.py

@@ -5,9 +5,10 @@ from http.server import HTTPServer, BaseHTTPRequestHandler
 from urllib.parse import urlencode, parse_qs
 from tools.lib.api import CommaApi, APIError
 from tools.lib.auth_config import set_token
+from typing import Dict, Any
 
 class ClientRedirectServer(HTTPServer):
-  query_params = {}
+  query_params: Dict[str, Any] = {}
 
 class ClientRedirectHandler(BaseHTTPRequestHandler):
   def do_GET(self):

tools/replay/lib/ui_helpers.py

@@ -1,15 +1,16 @@
 from collections import namedtuple
+from typing import Any, Dict, Tuple
 
 import matplotlib
 import matplotlib.pyplot as plt
 import numpy as np
 import pygame
 
-from tools.lib.lazy_property import lazy_property
-from selfdrive.config import UIParams as UP
 from selfdrive.config import RADAR_TO_CAMERA
+from selfdrive.config import UIParams as UP
 from selfdrive.controls.lib.lane_planner import (compute_path_pinv,
                                                  model_polyfit)
+from tools.lib.lazy_property import lazy_property
 
 RED = (255, 0, 0)
 GREEN = (0, 255, 0)
@@ -34,7 +35,7 @@ METER_WIDTH = 20
 
 ModelUIData = namedtuple("ModelUIData", ["cpath", "lpath", "rpath", "lead", "lead_future"])
 
-_COLOR_CACHE = {}
+_COLOR_CACHE : Dict[Tuple[int, int, int], Any] = {}
 def find_color(lidar_surface, color):
   if color in _COLOR_CACHE:
     return _COLOR_CACHE[color]

tools/replay/rqplot.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python
+# type: ignore
 import sys
 import matplotlib.pyplot as plt
 import numpy as np
@@ -78,4 +79,3 @@ if __name__ == "__main__":
 
     # just a bit of wait to avoid 100% CPU usage
     time.sleep(0.001)
-

tools/sim/bridge.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python3
+# type: ignore
 import time
 import math
 import atexit

tools/sim/lib/keyboard_ctrl.py

@@ -1,8 +1,9 @@
 import sys
 import termios
 import time
-from termios import (BRKINT, VMIN, CS8, CSIZE, ECHO, ICANON, ICRNL, IEXTEN,
-                     INPCK, ISIG, ISTRIP, IXON, PARENB, VTIME)
+from termios import (BRKINT, CS8, CSIZE, ECHO, ICANON, ICRNL, IEXTEN, INPCK,
+                     ISIG, ISTRIP, IXON, PARENB, VMIN, VTIME)
+from typing import Any
 
 # Indexes for termios list.
 IFLAG = 0
@@ -53,7 +54,7 @@ def test(q):
 
 if __name__ == '__main__':
   from multiprocessing import Process, Queue
-  q = Queue()
+  q : Any = Queue()
   p = Process(target=test, args=(q,))
   p.daemon = True
   p.start()