switch from mypy to ty (#36961)

system/athena/athenad.py

@@ -314,7 +314,7 @@ def upload_handler(end_event: threading.Event) -> None:
       cloudlog.exception("athena.upload_handler.exception")
 
 
-def _do_upload(upload_item: UploadItem, callback: Callable = None) -> requests.Response:
+def _do_upload(upload_item: UploadItem, callback: Callable | None = None) -> requests.Response:
   path = upload_item.path
   compress = False
 
@@ -805,7 +805,7 @@ def backoff(retries: int) -> int:
   return random.randrange(0, min(128, int(2 ** retries)))
 
 
-def main(exit_event: threading.Event = None):
+def main(exit_event: threading.Event | None = None):
   try:
     set_core_affinity([0, 1, 2, 3])
   except Exception:

system/athena/tests/test_athenad.py

@@ -97,7 +97,7 @@ class TestAthenadMethods:
         break
 
   @staticmethod
-  def _create_file(file: str, parent: str = None, data: bytes = b'') -> str:
+  def _create_file(file: str, parent: str | None = None, data: bytes = b'') -> str:
     fn = os.path.join(Paths.log_root() if parent is None else parent, file)
     os.makedirs(os.path.dirname(fn), exist_ok=True)
     with open(fn, 'wb') as f:

system/loggerd/tests/loggerd_tests_common.py

@@ -10,7 +10,7 @@ from openpilot.system.hardware.hw import Paths
 from openpilot.system.loggerd.xattr_cache import setxattr
 
 
-def create_random_file(file_path: Path, size_mb: float, lock: bool = False, upload_xattr: bytes = None) -> None:
+def create_random_file(file_path: Path, size_mb: float, lock: bool = False, upload_xattr: bytes | None = None) -> None:
   file_path.parent.mkdir(parents=True, exist_ok=True)
 
   if lock:
@@ -80,7 +80,7 @@ class UploaderTestCase:
     self.params.put("DongleId", "0000000000000000")
 
   def make_file_with_data(self, f_dir: str, fn: str, size_mb: float = .1, lock: bool = False,
-                          upload_xattr: bytes = None, preserve_xattr: bytes = None) -> Path:
+                          upload_xattr: bytes | None = None, preserve_xattr: bytes | None = None) -> Path:
     file_path = Path(Paths.log_root()) / f_dir / fn
     create_random_file(file_path, size_mb, lock, upload_xattr)
 

system/loggerd/tests/test_uploader.py

@@ -50,7 +50,7 @@ class TestUploader(UploaderTestCase):
     self.end_event.set()
     self.up_thread.join()
 
-  def gen_files(self, lock=False, xattr: bytes = None, boot=True) -> list[Path]:
+  def gen_files(self, lock=False, xattr: bytes | None = None, boot=True) -> list[Path]:
     f_paths = []
     for t in ["qlog", "rlog", "dcamera.hevc", "fcamera.hevc"]:
       f_paths.append(self.make_file_with_data(self.seg_dir, t, 1, lock=lock, upload_xattr=xattr))

system/loggerd/uploader.py

@@ -226,7 +226,7 @@ class Uploader:
     return self.upload(name, key, fn, network_type, metered)
 
 
-def main(exit_event: threading.Event = None) -> None:
+def main(exit_event: threading.Event | None = None) -> None:
   if exit_event is None:
     exit_event = threading.Event()
 

system/manager/process.py

@@ -81,7 +81,7 @@ class ManagerProcess(ABC):
     self.stop(sig=signal.SIGKILL)
     self.start()
 
-  def stop(self, retry: bool = True, block: bool = True, sig: signal.Signals = None) -> int | None:
+  def stop(self, retry: bool = True, block: bool = True, sig: signal.Signals | None = None) -> int | None:
     if self.proc is None:
       return None
 

system/ui/mici_updater.py

@@ -102,14 +102,15 @@ class Updater(Widget):
     self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                     text=True, bufsize=1, universal_newlines=True)
 
-    for line in self.process.stdout:
-      parts = line.strip().split(":")
-      if len(parts) == 2:
-        self.progress_text = parts[0].lower()
-        try:
-          self.progress_value = int(float(parts[1]))
-        except ValueError:
-          pass
+    if self.process.stdout is not None:
+      for line in self.process.stdout:
+        parts = line.strip().split(":")
+        if len(parts) == 2:
+          self.progress_text = parts[0].lower()
+          try:
+            self.progress_value = int(float(parts[1]))
+          except ValueError:
+            pass
 
     exit_code = self.process.wait()
     if exit_code == 0:

system/ui/tici_updater.py

@@ -71,14 +71,15 @@ class Updater(Widget):
     self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                     text=True, bufsize=1, universal_newlines=True)
 
-    for line in self.process.stdout:
-      parts = line.strip().split(":")
-      if len(parts) == 2:
-        self.progress_text = parts[0]
-        try:
-          self.progress_value = int(float(parts[1]))
-        except ValueError:
-          pass
+    if self.process.stdout is not None:
+      for line in self.process.stdout:
+        parts = line.strip().split(":")
+        if len(parts) == 2:
+          self.progress_text = parts[0]
+          try:
+            self.progress_value = int(float(parts[1]))
+          except ValueError:
+            pass
 
     exit_code = self.process.wait()
     if exit_code == 0:

system/ui/widgets/__init__.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import abc
 import pyray as rl
 from enum import IntEnum
@@ -91,7 +93,7 @@ class Widget(abc.ABC):
       return self._rect
     return rl.get_collision_rec(self._rect, self._parent_rect)
 
-  def render(self, rect: rl.Rectangle = None) -> bool | int | None:
+  def render(self, rect: rl.Rectangle | None = None) -> bool | int | None:
     if rect is not None:
       self.set_rect(rect)
 
@@ -359,7 +361,7 @@ class NavWidget(Widget, abc.ABC):
 
     self.set_position(self._rect.x, new_y)
 
-  def render(self, rect: rl.Rectangle = None) -> bool | int | None:
+  def render(self, rect: rl.Rectangle | None = None) -> bool | int | None:
     ret = super().render(rect)
 
     if self.back_enabled:

system/ui/widgets/label.py

@@ -31,13 +31,13 @@ class MiciLabel(Widget):
   def __init__(self,
                text: str,
                font_size: int = DEFAULT_TEXT_SIZE,
-               width: int = None,
+               width: int | None = None,
                color: rl.Color = DEFAULT_TEXT_COLOR,
                font_weight: FontWeight = FontWeight.NORMAL,
                alignment: int = rl.GuiTextAlignment.TEXT_ALIGN_LEFT,
                alignment_vertical: int = rl.GuiTextAlignmentVertical.TEXT_ALIGN_TOP,
                spacing: int = 0,
-               line_height: int = None,
+               line_height: int | None = None,
                elide_right: bool = True,
                wrap_text: bool = False,
                scroll: bool = False):

system/ui/widgets/list_view.py

@@ -174,8 +174,8 @@ class TextAction(ItemAction):
 
 
 class DualButtonAction(ItemAction):
-  def __init__(self, left_text: str | Callable[[], str], right_text: str | Callable[[], str], left_callback: Callable = None,
-               right_callback: Callable = None, enabled: bool | Callable[[], bool] = True):
+  def __init__(self, left_text: str | Callable[[], str], right_text: str | Callable[[], str], left_callback: Callable | None = None,
+               right_callback: Callable | None = None, enabled: bool | Callable[[], bool] = True):
     super().__init__(width=0, enabled=enabled)  # Width 0 means use full width
     self.left_button = Button(left_text, click_callback=left_callback, button_style=ButtonStyle.NORMAL, text_padding=0)
     self.right_button = Button(right_text, click_callback=right_callback, button_style=ButtonStyle.DANGER, text_padding=0)
@@ -207,7 +207,7 @@ class DualButtonAction(ItemAction):
 
 
 class MultipleButtonAction(ItemAction):
-  def __init__(self, buttons: list[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable = None):
+  def __init__(self, buttons: list[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable | None = None):
     super().__init__(width=len(buttons) * button_width + (len(buttons) - 1) * RIGHT_ITEM_PADDING, enabled=True)
     self.buttons = buttons
     self.button_width = button_width
@@ -454,13 +454,14 @@ def text_item(title: str | Callable[[], str], value: str | Callable[[], str], de
   return ListItem(title=title, description=description, action_item=action, callback=callback)
 
 
-def dual_button_item(left_text: str | Callable[[], str], right_text: str | Callable[[], str], left_callback: Callable = None, right_callback: Callable = None,
+def dual_button_item(left_text: str | Callable[[], str], right_text: str | Callable[[], str],
+                     left_callback: Callable | None = None, right_callback: Callable | None = None,
                      description: str | Callable[[], str] | None = None, enabled: bool | Callable[[], bool] = True) -> ListItem:
   action = DualButtonAction(left_text, right_text, left_callback, right_callback, enabled)
   return ListItem(title="", description=description, action_item=action)
 
 
 def multiple_button_item(title: str | Callable[[], str], description: str | Callable[[], str], buttons: list[str | Callable[[], str]], selected_index: int,
-                         button_width: int = BUTTON_WIDTH, callback: Callable = None, icon: str = ""):
+                         button_width: int = BUTTON_WIDTH, callback: Callable | None = None, icon: str = ""):
   action = MultipleButtonAction(buttons, button_width, selected_index, callback=callback)
   return ListItem(title=title, description=description, icon=icon, action_item=action)

system/updated/casync/casync.py

@@ -168,7 +168,7 @@ def build_chunk_dict(chunks: list[Chunk]) -> ChunkDict:
 def extract(target: list[Chunk],
             sources: list[tuple[str, ChunkReader, ChunkDict]],
             out_path: str,
-            progress: Callable[[int], None] = None):
+            progress: Callable[[int], None] | None = None):
   stats: dict[str, int] = defaultdict(int)
 
   mode = 'rb+' if os.path.exists(out_path) else 'wb'
@@ -208,7 +208,7 @@ def extract_directory(target: list[Chunk],
             sources: list[tuple[str, ChunkReader, ChunkDict]],
             out_path: str,
             tmp_file: str,
-            progress: Callable[[int], None] = None):
+            progress: Callable[[int], None] | None = None):
   """extract a directory stored as a casync tar archive"""
 
   stats = extract(target, sources, tmp_file, progress)

system/updated/updated.py

@@ -68,7 +68,7 @@ def write_time_to_param(params, param) -> None:
   t = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
   params.put(param, t)
 
-def run(cmd: list[str], cwd: str = None) -> str:
+def run(cmd: list[str], cwd: str | None = None) -> str:
   return subprocess.check_output(cmd, cwd=cwd, stderr=subprocess.STDOUT, encoding='utf8')
 
 

system/webrtc/device/audio.py

@@ -16,7 +16,7 @@ class AudioInputStreamTrack(aiortc.mediastreams.AudioStreamTrack):
       pyaudio.paFloat32: 'flt',
   }
 
-  def __init__(self, audio_format: int = pyaudio.paInt16, rate: int = 16000, channels: int = 1, packet_time: float = 0.020, device_index: int = None):
+  def __init__(self, audio_format: int = pyaudio.paInt16, rate: int = 16000, channels: int = 1, packet_time: float = 0.020, device_index: int | None = None):
     super().__init__()
 
     self.p = pyaudio.PyAudio()
@@ -48,7 +48,7 @@ class AudioInputStreamTrack(aiortc.mediastreams.AudioStreamTrack):
 
 
 class AudioOutputSpeaker:
-  def __init__(self, audio_format: int = pyaudio.paInt16, rate: int = 48000, channels: int = 2, packet_time: float = 0.2, device_index: int = None):
+  def __init__(self, audio_format: int = pyaudio.paInt16, rate: int = 48000, channels: int = 2, packet_time: float = 0.2, device_index: int | None = None):
 
     chunk_size = int(packet_time * rate)
     self.p = pyaudio.PyAudio()