use pyupgrade to update to new typing syntax (#31580)

* add pyupgrade hook

* run pyupgrade (pre-commit run -a)

* ruff --fix

* Revert "add pyupgrade hook"

This reverts commit 56ec18b.

* revert changes to third_party/

* manual type fixes

* explicit Optional wrapping capnp objects
old-commit-hash: 995250a

common/file_helpers.py

@@ -1,7 +1,6 @@
 import os
 import tempfile
 import contextlib
-from typing import Optional
 
 
 class CallbackReader:
@@ -24,7 +23,7 @@ def read(self, *args, **kwargs):
 
 
 @contextlib.contextmanager
-def atomic_write_in_dir(path: str, mode: str = 'w', buffering: int = -1, encoding: Optional[str] = None, newline: Optional[str] = None,
+def atomic_write_in_dir(path: str, mode: str = 'w', buffering: int = -1, encoding: str | None = None, newline: str | None = None,
                         overwrite: bool = False):
   """Write to a file atomically using a temporary file in the same directory as the destination file."""
   dir_name = os.path.dirname(path)

common/gpio.py

@@ -1,6 +1,5 @@
 import os
 from functools import lru_cache
-from typing import Optional, List
 
 def gpio_init(pin: int, output: bool) -> None:
   try:
@@ -16,7 +15,7 @@ def gpio_set(pin: int, high: bool) -> None:
   except Exception as e:
     print(f"Failed to set gpio {pin} value: {e}")
 
-def gpio_read(pin: int) -> Optional[bool]:
+def gpio_read(pin: int) -> bool | None:
   val = None
   try:
     with open(f"/sys/class/gpio/gpio{pin}/value", 'rb') as f:
@@ -37,15 +36,15 @@ def gpio_export(pin: int) -> None:
     print(f"Failed to export gpio {pin}")
 
 @lru_cache(maxsize=None)
-def get_irq_action(irq: int) -> List[str]:
+def get_irq_action(irq: int) -> list[str]:
   try:
     with open(f"/sys/kernel/irq/{irq}/actions") as f:
       actions = f.read().strip().split(',')
       return actions
   except FileNotFoundError:
     return []
 
-def get_irqs_for_action(action: str) -> List[str]:
+def get_irqs_for_action(action: str) -> list[str]:
   ret = []
   with open("/proc/interrupts") as f:
     for l in f.readlines():

common/mock/__init__.py

@@ -6,7 +6,6 @@
 
 import functools
 import threading
-from typing import List, Union
 from cereal.messaging import PubMaster
 from cereal.services import SERVICE_LIST
 from openpilot.common.mock.generators import generate_liveLocationKalman
@@ -18,7 +17,7 @@
 }
 
 
-def generate_messages_loop(services: List[str], done: threading.Event):
+def generate_messages_loop(services: list[str], done: threading.Event):
   pm = PubMaster(services)
   rk = Ratekeeper(100)
   i = 0
@@ -32,7 +31,7 @@ def generate_messages_loop(services: List[str], done: threading.Event):
     rk.keep_time()
 
 
-def mock_messages(services: Union[List[str], str]):
+def mock_messages(services: list[str] | str):
   if isinstance(services, str):
     services = [services]
 

common/prefix.py

@@ -2,14 +2,13 @@
 import shutil
 import uuid
 
-from typing import Optional
 
 from openpilot.common.params import Params
 from openpilot.system.hardware.hw import Paths
 from openpilot.system.hardware.hw import DEFAULT_DOWNLOAD_CACHE_ROOT
 
 class OpenpilotPrefix:
-  def __init__(self, prefix: Optional[str] = None, clean_dirs_on_exit: bool = True, shared_download_cache: bool = False):
+  def __init__(self, prefix: str | None = None, clean_dirs_on_exit: bool = True, shared_download_cache: bool = False):
     self.prefix = prefix if prefix else str(uuid.uuid4().hex[0:15])
     self.msgq_path = os.path.join('/dev/shm', self.prefix)
     self.clean_dirs_on_exit = clean_dirs_on_exit

common/realtime.py

@@ -3,7 +3,6 @@
 import os
 import time
 from collections import deque
-from typing import Optional, List, Union
 
 from setproctitle import getproctitle
 
@@ -33,20 +32,20 @@ def set_realtime_priority(level: int) -> None:
     os.sched_setscheduler(0, os.SCHED_FIFO, os.sched_param(level))
 
 
-def set_core_affinity(cores: List[int]) -> None:
+def set_core_affinity(cores: list[int]) -> None:
   if not PC:
     os.sched_setaffinity(0, cores)
 
 
-def config_realtime_process(cores: Union[int, List[int]], priority: int) -> None:
+def config_realtime_process(cores: int | list[int], priority: int) -> None:
   gc.disable()
   set_realtime_priority(priority)
   c = cores if isinstance(cores, list) else [cores, ]
   set_core_affinity(c)
 
 
 class Ratekeeper:
-  def __init__(self, rate: float, print_delay_threshold: Optional[float] = 0.0) -> None:
+  def __init__(self, rate: float, print_delay_threshold: float | None = 0.0) -> None:
     """Rate in Hz for ratekeeping. print_delay_threshold must be nonnegative."""
     self._interval = 1. / rate
     self._next_frame_time = time.monotonic() + self._interval

common/transformations/orientation.py

@@ -1,5 +1,5 @@
 import numpy as np
-from typing import Callable
+from collections.abc import Callable
 
 from openpilot.common.transformations.transformations import (ecef_euler_from_ned_single,
                                                     euler2quat_single,

openpilot/__init__.py

@@ -1,3 +0,0 @@
-
-
-

selfdrive/athena/athenad.py

@@ -19,7 +19,8 @@
 from datetime import datetime
 from functools import partial
 from queue import Queue
-from typing import Callable, Dict, List, Optional, Set, Union, cast
+from typing import cast
+from collections.abc import Callable
 
 import requests
 from jsonrpc import JSONRPCResponseManager, dispatcher
@@ -55,17 +56,17 @@
 
 NetworkType = log.DeviceState.NetworkType
 
-UploadFileDict = Dict[str, Union[str, int, float, bool]]
-UploadItemDict = Dict[str, Union[str, bool, int, float, Dict[str, str]]]
+UploadFileDict = dict[str, str | int | float | bool]
+UploadItemDict = dict[str, str | bool | int | float | dict[str, str]]
 
-UploadFilesToUrlResponse = Dict[str, Union[int, List[UploadItemDict], List[str]]]
+UploadFilesToUrlResponse = dict[str, int | list[UploadItemDict] | list[str]]
 
 
 @dataclass
 class UploadFile:
   fn: str
   url: str
-  headers: Dict[str, str]
+  headers: dict[str, str]
   allow_cellular: bool
 
   @classmethod
@@ -77,9 +78,9 @@ def from_dict(cls, d: dict) -> UploadFile:
 class UploadItem:
   path: str
   url: str
-  headers: Dict[str, str]
+  headers: dict[str, str]
   created_at: int
-  id: Optional[str]
+  id: str | None
   retry_count: int = 0
   current: bool = False
   progress: float = 0
@@ -97,9 +98,9 @@ def from_dict(cls, d: dict) -> UploadItem:
 upload_queue: Queue[UploadItem] = queue.Queue()
 low_priority_send_queue: Queue[str] = queue.Queue()
 log_recv_queue: Queue[str] = queue.Queue()
-cancelled_uploads: Set[str] = set()
+cancelled_uploads: set[str] = set()
 
-cur_upload_items: Dict[int, Optional[UploadItem]] = {}
+cur_upload_items: dict[int, UploadItem | None] = {}
 
 
 def strip_bz2_extension(fn: str) -> str:
@@ -127,14 +128,14 @@ def initialize(upload_queue: Queue[UploadItem]) -> None:
   @staticmethod
   def cache(upload_queue: Queue[UploadItem]) -> None:
     try:
-      queue: List[Optional[UploadItem]] = list(upload_queue.queue)
+      queue: list[UploadItem | None] = list(upload_queue.queue)
       items = [asdict(i) for i in queue if i is not None and (i.id not in cancelled_uploads)]
       Params().put("AthenadUploadQueue", json.dumps(items))
     except Exception:
       cloudlog.exception("athena.UploadQueueCache.cache.exception")
 
 
-def handle_long_poll(ws: WebSocket, exit_event: Optional[threading.Event]) -> None:
+def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None:
   end_event = threading.Event()
 
   threads = [
@@ -278,7 +279,7 @@ def upload_handler(end_event: threading.Event) -> None:
       cloudlog.exception("athena.upload_handler.exception")
 
 
-def _do_upload(upload_item: UploadItem, callback: Optional[Callable] = None) -> requests.Response:
+def _do_upload(upload_item: UploadItem, callback: Callable | None = None) -> requests.Response:
   path = upload_item.path
   compress = False
 
@@ -317,7 +318,7 @@ def getMessage(service: str, timeout: int = 1000) -> dict:
 
 
 @dispatcher.add_method
-def getVersion() -> Dict[str, str]:
+def getVersion() -> dict[str, str]:
   return {
     "version": get_version(),
     "remote": get_normalized_origin(),
@@ -327,7 +328,7 @@ def getVersion() -> Dict[str, str]:
 
 
 @dispatcher.add_method
-def setNavDestination(latitude: int = 0, longitude: int = 0, place_name: Optional[str] = None, place_details: Optional[str] = None) -> Dict[str, int]:
+def setNavDestination(latitude: int = 0, longitude: int = 0, place_name: str | None = None, place_details: str | None = None) -> dict[str, int]:
   destination = {
     "latitude": latitude,
     "longitude": longitude,
@@ -339,7 +340,7 @@ def setNavDestination(latitude: int = 0, longitude: int = 0, place_name: Optiona
   return {"success": 1}
 
 
-def scan_dir(path: str, prefix: str) -> List[str]:
+def scan_dir(path: str, prefix: str) -> list[str]:
   files = []
   # only walk directories that match the prefix
   # (glob and friends traverse entire dir tree)
@@ -359,12 +360,12 @@ def scan_dir(path: str, prefix: str) -> List[str]:
   return files
 
 @dispatcher.add_method
-def listDataDirectory(prefix='') -> List[str]:
+def listDataDirectory(prefix='') -> list[str]:
   return scan_dir(Paths.log_root(), prefix)
 
 
 @dispatcher.add_method
-def uploadFileToUrl(fn: str, url: str, headers: Dict[str, str]) -> UploadFilesToUrlResponse:
+def uploadFileToUrl(fn: str, url: str, headers: dict[str, str]) -> UploadFilesToUrlResponse:
   # this is because mypy doesn't understand that the decorator doesn't change the return type
   response: UploadFilesToUrlResponse = uploadFilesToUrls([{
     "fn": fn,
@@ -375,11 +376,11 @@ def uploadFileToUrl(fn: str, url: str, headers: Dict[str, str]) -> UploadFilesTo
 
 
 @dispatcher.add_method
-def uploadFilesToUrls(files_data: List[UploadFileDict]) -> UploadFilesToUrlResponse:
+def uploadFilesToUrls(files_data: list[UploadFileDict]) -> UploadFilesToUrlResponse:
   files = map(UploadFile.from_dict, files_data)
 
-  items: List[UploadItemDict] = []
-  failed: List[str] = []
+  items: list[UploadItemDict] = []
+  failed: list[str] = []
   for file in files:
     if len(file.fn) == 0 or file.fn[0] == '/' or '..' in file.fn or len(file.url) == 0:
       failed.append(file.fn)
@@ -418,13 +419,13 @@ def uploadFilesToUrls(files_data: List[UploadFileDict]) -> UploadFilesToUrlRespo
 
 
 @dispatcher.add_method
-def listUploadQueue() -> List[UploadItemDict]:
+def listUploadQueue() -> list[UploadItemDict]:
   items = list(upload_queue.queue) + list(cur_upload_items.values())
   return [asdict(i) for i in items if (i is not None) and (i.id not in cancelled_uploads)]
 
 
 @dispatcher.add_method
-def cancelUpload(upload_id: Union[str, List[str]]) -> Dict[str, Union[int, str]]:
+def cancelUpload(upload_id: str | list[str]) -> dict[str, int | str]:
   if not isinstance(upload_id, list):
     upload_id = [upload_id]
 
@@ -437,7 +438,7 @@ def cancelUpload(upload_id: Union[str, List[str]]) -> Dict[str, Union[int, str]]
   return {"success": 1}
 
 @dispatcher.add_method
-def setRouteViewed(route: str) -> Dict[str, Union[int, str]]:
+def setRouteViewed(route: str) -> dict[str, int | str]:
   # maintain a list of the last 10 routes viewed in connect
   params = Params()
 
@@ -452,7 +453,7 @@ def setRouteViewed(route: str) -> Dict[str, Union[int, str]]:
   return {"success": 1}
 
 
-def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local_port: int) -> Dict[str, int]:
+def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local_port: int) -> dict[str, int]:
   try:
     if local_port not in LOCAL_PORT_WHITELIST:
       raise Exception("Requested local port not whitelisted")
@@ -486,7 +487,7 @@ def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local
 
 
 @dispatcher.add_method
-def getPublicKey() -> Optional[str]:
+def getPublicKey() -> str | None:
   if not os.path.isfile(Paths.persist_root() + '/comma/id_rsa.pub'):
     return None
 
@@ -526,7 +527,7 @@ def getNetworks():
 
 
 @dispatcher.add_method
-def takeSnapshot() -> Optional[Union[str, Dict[str, str]]]:
+def takeSnapshot() -> str | dict[str, str] | None:
   from openpilot.system.camerad.snapshot.snapshot import jpeg_write, snapshot
   ret = snapshot()
   if ret is not None:
@@ -543,7 +544,7 @@ def b64jpeg(x):
     raise Exception("not available while camerad is started")
 
 
-def get_logs_to_send_sorted() -> List[str]:
+def get_logs_to_send_sorted() -> list[str]:
   # TODO: scan once then use inotify to detect file creation/deletion
   curr_time = int(time.time())
   logs = []
@@ -766,7 +767,7 @@ def backoff(retries: int) -> int:
   return random.randrange(0, min(128, int(2 ** retries)))
 
 
-def main(exit_event: Optional[threading.Event] = None):
+def main(exit_event: threading.Event | None = None):
   try:
     set_core_affinity([0, 1, 2, 3])
   except Exception:

selfdrive/athena/registration.py

@@ -3,7 +3,6 @@
 import json
 import jwt
 from pathlib import Path
-from typing import Optional
 
 from datetime import datetime, timedelta
 from openpilot.common.api import api_get
@@ -23,12 +22,12 @@ def is_registered_device() -> bool:
   return dongle not in (None, UNREGISTERED_DONGLE_ID)
 
 
-def register(show_spinner=False) -> Optional[str]:
+def register(show_spinner=False) -> str | None:
   params = Params()
 
   IMEI = params.get("IMEI", encoding='utf8')
   HardwareSerial = params.get("HardwareSerial", encoding='utf8')
-  dongle_id: Optional[str] = params.get("DongleId", encoding='utf8')
+  dongle_id: str | None = params.get("DongleId", encoding='utf8')
   needs_registration = None in (IMEI, HardwareSerial, dongle_id)
 
   pubkey = Path(Paths.persist_root()+"/comma/id_rsa.pub")
@@ -48,8 +47,8 @@ def register(show_spinner=False) -> Optional[str]:
     # Block until we get the imei
     serial = HARDWARE.get_serial()
     start_time = time.monotonic()
-    imei1: Optional[str] = None
-    imei2: Optional[str] = None
+    imei1: str | None = None
+    imei2: str | None = None
     while imei1 is None and imei2 is None:
       try:
         imei1, imei2 = HARDWARE.get_imei(0), HARDWARE.get_imei(1)

selfdrive/athena/tests/test_athenad.py

@@ -12,7 +12,6 @@
 from dataclasses import asdict, replace
 from datetime import datetime, timedelta
 from parameterized import parameterized
-from typing import Optional
 
 from unittest import mock
 from websocket import ABNF
@@ -97,7 +96,7 @@ def _wait_for_upload():
         break
 
   @staticmethod
-  def _create_file(file: str, parent: Optional[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: