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 56ec18bb6b.

* revert changes to third_party/

* manual type fixes

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

common/file_helpers.py

@@ -1,7 +1,6 @@
 import os
 import tempfile
 import contextlib
-from typing import Optional
 
 
 class CallbackReader:
@@ -24,7 +23,7 @@ class CallbackReader:
 
 
 @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,7 +36,7 @@ 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(',')
@@ -45,7 +44,7 @@ def get_irq_action(irq: int) -> List[str]:
   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 @@ example in common/tests/test_mock.py
 
 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 @@ MOCK_GENERATOR = {
 }
 
 
-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 os
 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 gc
 import os
 import time
 from collections import deque
-from typing import Optional, List, Union
 
 from setproctitle import getproctitle
 
@@ -33,12 +32,12 @@ 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, ]
@@ -46,7 +45,7 @@ def config_realtime_process(cores: Union[int, List[int]], priority: int) -> None
 
 
 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 dataclasses import asdict, dataclass, replace
 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 @@ WS_FRAME_SIZE = 4096
 
 NetworkType = log.DeviceState.NetworkType
 
-UploadFileDict = Dict[str, Union[str, int, float, bool]]
+UploadFileDict = dict[str, str | int | float | bool]
-UploadItemDict = Dict[str, Union[str, bool, int, float, Dict[str, str]]]
+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 @@ class 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 @@ send_queue: Queue[str] = queue.Queue()
 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 @@ class UploadQueueCache:
   @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] = []
+  items: list[UploadItemDict] = []
-  failed: List[str] = []
+  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 takeSnapshot() -> Optional[Union[str, Dict[str, str]]]:
     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 time
 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
+    imei1: str | None = None
-    imei2: Optional[str] = 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 @@ import unittest
 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 @@ class TestAthenadMethods(unittest.TestCase):
         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:

selfdrive/athena/tests/test_athenad_ping.py

@@ -3,7 +3,7 @@ import subprocess
 import threading
 import time
 import unittest
-from typing import cast, Optional
+from typing import cast
 from unittest import mock
 
 from openpilot.common.params import Params
@@ -29,8 +29,8 @@ class TestAthenadPing(unittest.TestCase):
   athenad: threading.Thread
   exit_event: threading.Event
 
-  def _get_ping_time(self) -> Optional[str]:
+  def _get_ping_time(self) -> str | None:
-    return cast(Optional[str], self.params.get("LastAthenaPingTime", encoding="utf-8"))
+    return cast(str | None, self.params.get("LastAthenaPingTime", encoding="utf-8"))
 
   def _clear_ping_time(self) -> None:
     self.params.remove("LastAthenaPingTime")

selfdrive/boardd/pandad.py

@@ -4,7 +4,7 @@ import os
 import usb1
 import time
 import subprocess
-from typing import List, NoReturn
+from typing import NoReturn
 from functools import cmp_to_key
 
 from panda import Panda, PandaDFU, PandaProtocolMismatch, FW_PATH
@@ -124,7 +124,7 @@ def main() -> NoReturn:
       cloudlog.info(f"{len(panda_serials)} panda(s) found, connecting - {panda_serials}")
 
       # Flash pandas
-      pandas: List[Panda] = []
+      pandas: list[Panda] = []
       for serial in panda_serials:
         pandas.append(flash_panda(serial))
 

selfdrive/car/__init__.py

@@ -2,7 +2,6 @@
 from collections import namedtuple
 from dataclasses import dataclass
 from enum import ReprEnum
-from typing import Dict, List, Optional, Union
 
 import capnp
 
@@ -27,9 +26,9 @@ def apply_hysteresis(val: float, val_steady: float, hyst_gap: float) -> float:
   return val_steady
 
 
-def create_button_events(cur_btn: int, prev_btn: int, buttons_dict: Dict[int, capnp.lib.capnp._EnumModule],
+def create_button_events(cur_btn: int, prev_btn: int, buttons_dict: dict[int, capnp.lib.capnp._EnumModule],
-                         unpressed_btn: int = 0) -> List[capnp.lib.capnp._DynamicStructBuilder]:
+                         unpressed_btn: int = 0) -> list[capnp.lib.capnp._DynamicStructBuilder]:
-  events: List[capnp.lib.capnp._DynamicStructBuilder] = []
+  events: list[capnp.lib.capnp._DynamicStructBuilder] = []
 
   if cur_btn == prev_btn:
     return events
@@ -76,7 +75,7 @@ def scale_tire_stiffness(mass, wheelbase, center_to_front, tire_stiffness_factor
   return tire_stiffness_front, tire_stiffness_rear
 
 
-DbcDict = Dict[str, str]
+DbcDict = dict[str, str]
 
 
 def dbc_dict(pt_dbc, radar_dbc, chassis_dbc=None, body_dbc=None) -> DbcDict:
@@ -214,7 +213,7 @@ def get_safety_config(safety_model, safety_param = None):
 class CanBusBase:
   offset: int
 
-  def __init__(self, CP, fingerprint: Optional[Dict[int, Dict[int, int]]]) -> None:
+  def __init__(self, CP, fingerprint: dict[int, dict[int, int]] | None) -> None:
     if CP is None:
       assert fingerprint is not None
       num = max([k for k, v in fingerprint.items() if len(v)], default=0) // 4 + 1
@@ -244,7 +243,7 @@ class CanSignalRateCalculator:
     return self.rate
 
 
-CarInfos = Union[CarInfo, List[CarInfo]]
+CarInfos = CarInfo | list[CarInfo]
 
 
 @dataclass
@@ -260,7 +259,7 @@ class PlatformConfig:
   car_info: CarInfos
   dbc_dict: DbcDict
 
-  specs: Optional[CarSpecs] = None
+  specs: CarSpecs | None = None
 
   def __hash__(self) -> int:
     return hash(self.platform_str)
@@ -276,9 +275,9 @@ class Platforms(str, ReprEnum):
     return member
 
   @classmethod
-  def create_dbc_map(cls) -> Dict[str, DbcDict]:
+  def create_dbc_map(cls) -> dict[str, DbcDict]:
     return {p: p.config.dbc_dict for p in cls}
 
   @classmethod
-  def create_carinfo_map(cls) -> Dict[str, CarInfos]:
+  def create_carinfo_map(cls) -> dict[str, CarInfos]:
     return {p: p.config.car_info for p in cls}

selfdrive/car/car_helpers.py

@@ -1,6 +1,6 @@
 import os
 import time
-from typing import Callable, Dict, List, Optional, Tuple
+from collections.abc import Callable
 
 from cereal import car
 from openpilot.common.params import Params
@@ -63,7 +63,7 @@ def load_interfaces(brand_names):
   return ret
 
 
-def _get_interface_names() -> Dict[str, List[str]]:
+def _get_interface_names() -> dict[str, list[str]]:
   # returns a dict of brand name and its respective models
   brand_names = {}
   for brand_name, brand_models in get_interface_attr("CAR").items():
@@ -77,7 +77,7 @@ interface_names = _get_interface_names()
 interfaces = load_interfaces(interface_names)
 
 
-def can_fingerprint(next_can: Callable) -> Tuple[Optional[str], Dict[int, dict]]:
+def can_fingerprint(next_can: Callable) -> tuple[str | None, dict[int, dict]]:
   finger = gen_empty_fingerprint()
   candidate_cars = {i: all_legacy_fingerprint_cars() for i in [0, 1]}  # attempt fingerprint on both bus 0 and 1
   frame = 0

selfdrive/car/chrysler/values.py

@@ -1,6 +1,5 @@
 from enum import IntFlag, StrEnum
 from dataclasses import dataclass, field
-from typing import Dict, List, Optional, Union
 
 from cereal import car
 from panda.python import uds
@@ -66,7 +65,7 @@ class ChryslerCarInfo(CarInfo):
   car_parts: CarParts = field(default_factory=CarParts.common([CarHarness.fca]))
 
 
-CAR_INFO: Dict[str, Optional[Union[ChryslerCarInfo, List[ChryslerCarInfo]]]] = {
+CAR_INFO: dict[str, ChryslerCarInfo | list[ChryslerCarInfo] | None] = {
   CAR.PACIFICA_2017_HYBRID: ChryslerCarInfo("Chrysler Pacifica Hybrid 2017"),
   CAR.PACIFICA_2018_HYBRID: ChryslerCarInfo("Chrysler Pacifica Hybrid 2018"),
   CAR.PACIFICA_2019_HYBRID: ChryslerCarInfo("Chrysler Pacifica Hybrid 2019-23"),

selfdrive/car/docs.py

@@ -5,7 +5,6 @@ import jinja2
 import os
 from enum import Enum
 from natsort import natsorted
-from typing import Dict, List
 
 from cereal import car
 from openpilot.common.basedir import BASEDIR
@@ -14,7 +13,7 @@ from openpilot.selfdrive.car.docs_definitions import CarInfo, Column, CommonFoot
 from openpilot.selfdrive.car.car_helpers import interfaces, get_interface_attr
 
 
-def get_all_footnotes() -> Dict[Enum, int]:
+def get_all_footnotes() -> dict[Enum, int]:
   all_footnotes = list(CommonFootnote)
   for footnotes in get_interface_attr("Footnote", ignore_none=True).values():
     all_footnotes.extend(footnotes)
@@ -25,8 +24,8 @@ CARS_MD_OUT = os.path.join(BASEDIR, "docs", "CARS.md")
 CARS_MD_TEMPLATE = os.path.join(BASEDIR, "selfdrive", "car", "CARS_template.md")
 
 
-def get_all_car_info() -> List[CarInfo]:
+def get_all_car_info() -> list[CarInfo]:
-  all_car_info: List[CarInfo] = []
+  all_car_info: list[CarInfo] = []
   footnotes = get_all_footnotes()
   for model, car_info in get_interface_attr("CAR_INFO", combine_brands=True).items():
     # If available, uses experimental longitudinal limits for the docs
@@ -47,19 +46,19 @@ def get_all_car_info() -> List[CarInfo]:
       all_car_info.append(_car_info)
 
   # Sort cars by make and model + year
-  sorted_cars: List[CarInfo] = natsorted(all_car_info, key=lambda car: car.name.lower())
+  sorted_cars: list[CarInfo] = natsorted(all_car_info, key=lambda car: car.name.lower())
   return sorted_cars
 
 
-def group_by_make(all_car_info: List[CarInfo]) -> Dict[str, List[CarInfo]]:
+def group_by_make(all_car_info: list[CarInfo]) -> dict[str, list[CarInfo]]:
   sorted_car_info = defaultdict(list)
   for car_info in all_car_info:
     sorted_car_info[car_info.make].append(car_info)
   return dict(sorted_car_info)
 
 
-def generate_cars_md(all_car_info: List[CarInfo], template_fn: str) -> str:
+def generate_cars_md(all_car_info: list[CarInfo], template_fn: str) -> str:
-  with open(template_fn, "r") as f:
+  with open(template_fn) as f:
     template = jinja2.Template(f.read(), trim_blocks=True, lstrip_blocks=True)
 
   footnotes = [fn.value.text for fn in get_all_footnotes()]

selfdrive/car/docs_definitions.py

@@ -3,7 +3,6 @@ from collections import namedtuple
 import copy
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Dict, List, Optional, Tuple, Union
 
 from cereal import car
 from openpilot.common.conversions import Conversions as CV
@@ -35,7 +34,7 @@ class Star(Enum):
 @dataclass
 class BasePart:
   name: str
-  parts: List[Enum] = field(default_factory=list)
+  parts: list[Enum] = field(default_factory=list)
 
   def all_parts(self):
     # Recursively get all parts
@@ -76,7 +75,7 @@ class Accessory(EnumBase):
 
 @dataclass
 class BaseCarHarness(BasePart):
-  parts: List[Enum] = field(default_factory=lambda: [Accessory.harness_box, Accessory.comma_power_v2, Cable.rj45_cable_7ft])
+  parts: list[Enum] = field(default_factory=lambda: [Accessory.harness_box, Accessory.comma_power_v2, Cable.rj45_cable_7ft])
   has_connector: bool = True  # without are hidden on the harness connector page
 
 
@@ -149,18 +148,18 @@ class PartType(Enum):
   tool = Tool
 
 
-DEFAULT_CAR_PARTS: List[EnumBase] = [Device.threex]
+DEFAULT_CAR_PARTS: list[EnumBase] = [Device.threex]
 
 
 @dataclass
 class CarParts:
-  parts: List[EnumBase] = field(default_factory=list)
+  parts: list[EnumBase] = field(default_factory=list)
 
   def __call__(self):
     return copy.deepcopy(self)
 
   @classmethod
-  def common(cls, add: Optional[List[EnumBase]] = None, remove: Optional[List[EnumBase]] = None):
+  def common(cls, add: list[EnumBase] | None = None, remove: list[EnumBase] | None = None):
     p = [part for part in (add or []) + DEFAULT_CAR_PARTS if part not in (remove or [])]
     return cls(p)
 
@@ -186,7 +185,7 @@ class CommonFootnote(Enum):
     Column.LONGITUDINAL)
 
 
-def get_footnotes(footnotes: List[Enum], column: Column) -> List[Enum]:
+def get_footnotes(footnotes: list[Enum], column: Column) -> list[Enum]:
   # Returns applicable footnotes given current column
   return [fn for fn in footnotes if fn.value.column == column]
 
@@ -209,7 +208,7 @@ def get_year_list(years):
   return years_list
 
 
-def split_name(name: str) -> Tuple[str, str, str]:
+def split_name(name: str) -> tuple[str, str, str]:
   make, model = name.split(" ", 1)
   years = ""
   match = re.search(MODEL_YEARS_RE, model)
@@ -233,13 +232,13 @@ class CarInfo:
 
   # the minimum compatibility requirements for this model, regardless
   # of market. can be a package, trim, or list of features
-  requirements: Optional[str] = None
+  requirements: str | None = None
 
-  video_link: Optional[str] = None
+  video_link: str | None = None
-  footnotes: List[Enum] = field(default_factory=list)
+  footnotes: list[Enum] = field(default_factory=list)
-  min_steer_speed: Optional[float] = None
+  min_steer_speed: float | None = None
-  min_enable_speed: Optional[float] = None
+  min_enable_speed: float | None = None
-  auto_resume: Optional[bool] = None
+  auto_resume: bool | None = None
 
   # all the parts needed for the supported car
   car_parts: CarParts = field(default_factory=CarParts)
@@ -248,7 +247,7 @@ class CarInfo:
     self.make, self.model, self.years = split_name(self.name)
     self.year_list = get_year_list(self.years)
 
-  def init(self, CP: car.CarParams, all_footnotes: Dict[Enum, int]):
+  def init(self, CP: car.CarParams, all_footnotes: dict[Enum, int]):
     self.car_name = CP.carName
     self.car_fingerprint = CP.carFingerprint
 
@@ -293,7 +292,7 @@ class CarInfo:
       if len(tools_docs):
         hardware_col += f'<details><summary>Tools</summary><sub>{display_func(tools_docs)}</sub></details>'
 
-    self.row: Dict[Enum, Union[str, Star]] = {
+    self.row: dict[Enum, str | Star] = {
       Column.MAKE: self.make,
       Column.MODEL: self.model,
       Column.PACKAGE: self.package,
@@ -352,7 +351,7 @@ class CarInfo:
         raise Exception(f"This notCar does not have a detail sentence: {CP.carFingerprint}")
 
   def get_column(self, column: Column, star_icon: str, video_icon: str, footnote_tag: str) -> str:
-    item: Union[str, Star] = self.row[column]
+    item: str | Star = self.row[column]
     if isinstance(item, Star):
       item = star_icon.format(item.value)
     elif column == Column.MODEL and len(self.years):

selfdrive/car/ecu_addrs.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 import capnp
 import time
-from typing import Optional, Set
 
 import cereal.messaging as messaging
 from panda.python.uds import SERVICE_TYPE
@@ -20,7 +19,7 @@ def make_tester_present_msg(addr, bus, subaddr=None):
   return make_can_msg(addr, bytes(dat), bus)
 
 
-def is_tester_present_response(msg: capnp.lib.capnp._DynamicStructReader, subaddr: Optional[int] = None) -> bool:
+def is_tester_present_response(msg: capnp.lib.capnp._DynamicStructReader, subaddr: int | None = None) -> bool:
   # ISO-TP messages are always padded to 8 bytes
   # tester present response is always a single frame
   dat_offset = 1 if subaddr is not None else 0
@@ -34,16 +33,16 @@ def is_tester_present_response(msg: capnp.lib.capnp._DynamicStructReader, subadd
   return False
 
 
-def get_all_ecu_addrs(logcan: messaging.SubSocket, sendcan: messaging.PubSocket, bus: int, timeout: float = 1, debug: bool = True) -> Set[EcuAddrBusType]:
+def get_all_ecu_addrs(logcan: messaging.SubSocket, sendcan: messaging.PubSocket, bus: int, timeout: float = 1, debug: bool = True) -> set[EcuAddrBusType]:
   addr_list = [0x700 + i for i in range(256)] + [0x18da00f1 + (i << 8) for i in range(256)]
-  queries: Set[EcuAddrBusType] = {(addr, None, bus) for addr in addr_list}
+  queries: set[EcuAddrBusType] = {(addr, None, bus) for addr in addr_list}
   responses = queries
   return get_ecu_addrs(logcan, sendcan, queries, responses, timeout=timeout, debug=debug)
 
 
-def get_ecu_addrs(logcan: messaging.SubSocket, sendcan: messaging.PubSocket, queries: Set[EcuAddrBusType],
+def get_ecu_addrs(logcan: messaging.SubSocket, sendcan: messaging.PubSocket, queries: set[EcuAddrBusType],
-                  responses: Set[EcuAddrBusType], timeout: float = 1, debug: bool = False) -> Set[EcuAddrBusType]:
+                  responses: set[EcuAddrBusType], timeout: float = 1, debug: bool = False) -> set[EcuAddrBusType]:
-  ecu_responses: Set[EcuAddrBusType] = set()  # set((addr, subaddr, bus),)
+  ecu_responses: set[EcuAddrBusType] = set()  # set((addr, subaddr, bus),)
   try:
     msgs = [make_tester_present_msg(addr, bus, subaddr) for addr, subaddr, bus in queries]
 

selfdrive/car/ford/tests/test_ford.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 import unittest
 from parameterized import parameterized
-from typing import Dict, Iterable, Optional, Tuple
+from collections.abc import Iterable
 
 import capnp
 
@@ -50,7 +50,7 @@ class TestFordFW(unittest.TestCase):
       self.assertIsNone(subaddr, "Unexpected ECU subaddress")
 
   @parameterized.expand(FW_VERSIONS.items())
-  def test_fw_versions(self, car_model: str, fw_versions: Dict[Tuple[capnp.lib.capnp._EnumModule, int, Optional[int]], Iterable[bytes]]):
+  def test_fw_versions(self, car_model: str, fw_versions: dict[tuple[capnp.lib.capnp._EnumModule, int, int | None], Iterable[bytes]]):
     for (ecu, addr, subaddr), fws in fw_versions.items():
       self.assertIn(ecu, ECU_FW_CORE, "Unexpected ECU")
       self.assertEqual(addr, ECU_ADDRESSES[ecu], "ECU address mismatch")

selfdrive/car/ford/values.py

@@ -1,7 +1,6 @@
 from collections import defaultdict
 from dataclasses import dataclass
 from enum import Enum, StrEnum
-from typing import Dict, List, Union
 
 from cereal import car
 from openpilot.selfdrive.car import AngleRateLimit, dbc_dict
@@ -59,7 +58,7 @@ class RADAR:
   DELPHI_MRR = 'FORD_CADS'
 
 
-DBC: Dict[str, Dict[str, str]] = defaultdict(lambda: dbc_dict("ford_lincoln_base_pt", RADAR.DELPHI_MRR))
+DBC: dict[str, dict[str, str]] = defaultdict(lambda: dbc_dict("ford_lincoln_base_pt", RADAR.DELPHI_MRR))
 
 # F-150 radar is not yet supported
 DBC[CAR.F_150_MK14] = dbc_dict("ford_lincoln_base_pt", None)
@@ -87,7 +86,7 @@ class FordCarInfo(CarInfo):
       self.car_parts = CarParts([Device.threex, harness])
 
 
-CAR_INFO: Dict[str, Union[CarInfo, List[CarInfo]]] = {
+CAR_INFO: dict[str, CarInfo | list[CarInfo]] = {
   CAR.BRONCO_SPORT_MK1: FordCarInfo("Ford Bronco Sport 2021-22"),
   CAR.ESCAPE_MK4: [
     FordCarInfo("Ford Escape 2020-22"),

selfdrive/car/fw_query_definitions.py

@@ -3,16 +3,16 @@ import capnp
 import copy
 from dataclasses import dataclass, field
 import struct
-from typing import Callable, Dict, List, Optional, Set, Tuple
+from collections.abc import Callable
 
 import panda.python.uds as uds
 
-AddrType = Tuple[int, Optional[int]]
+AddrType = tuple[int, int | None]
-EcuAddrBusType = Tuple[int, Optional[int], int]
+EcuAddrBusType = tuple[int, int | None, int]
-EcuAddrSubAddr = Tuple[int, int, Optional[int]]
+EcuAddrSubAddr = tuple[int, int, int | None]
 
-LiveFwVersions = Dict[AddrType, Set[bytes]]
+LiveFwVersions = dict[AddrType, set[bytes]]
-OfflineFwVersions = Dict[str, Dict[EcuAddrSubAddr, List[bytes]]]
+OfflineFwVersions = dict[str, dict[EcuAddrSubAddr, list[bytes]]]
 
 # A global list of addresses we will only ever consider for VIN responses
 # engine, hybrid controller, Ford abs, Hyundai CAN FD cluster, 29-bit engine, PGM-FI
@@ -71,9 +71,9 @@ class StdQueries:
 
 @dataclass
 class Request:
-  request: List[bytes]
+  request: list[bytes]
-  response: List[bytes]
+  response: list[bytes]
-  whitelist_ecus: List[int] = field(default_factory=list)
+  whitelist_ecus: list[int] = field(default_factory=list)
   rx_offset: int = 0x8
   bus: int = 1
   # Whether this query should be run on the first auxiliary panda (CAN FD cars for example)
@@ -86,15 +86,15 @@ class Request:
 
 @dataclass
 class FwQueryConfig:
-  requests: List[Request]
+  requests: list[Request]
   # TODO: make this automatic and remove hardcoded lists, or do fingerprinting with ecus
   # Overrides and removes from essential ecus for specific models and ecus (exact matching)
-  non_essential_ecus: Dict[capnp.lib.capnp._EnumModule, List[str]] = field(default_factory=dict)
+  non_essential_ecus: dict[capnp.lib.capnp._EnumModule, list[str]] = field(default_factory=dict)
   # Ecus added for data collection, not to be fingerprinted on
-  extra_ecus: List[Tuple[capnp.lib.capnp._EnumModule, int, Optional[int]]] = field(default_factory=list)
+  extra_ecus: list[tuple[capnp.lib.capnp._EnumModule, int, int | None]] = field(default_factory=list)
   # Function a brand can implement to provide better fuzzy matching. Takes in FW versions,
   # returns set of candidates. Only will match if one candidate is returned
-  match_fw_to_car_fuzzy: Optional[Callable[[LiveFwVersions, OfflineFwVersions], Set[str]]] = None
+  match_fw_to_car_fuzzy: Callable[[LiveFwVersions, OfflineFwVersions], set[str]] | None = None
 
   def __post_init__(self):
     for i in range(len(self.requests)):

selfdrive/car/fw_versions.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 from collections import defaultdict
-from typing import Any, DefaultDict, Dict, Iterator, List, Optional, Set, TypeVar
+from typing import Any, TypeVar
+from collections.abc import Iterator
 from tqdm import tqdm
 import capnp
 
@@ -27,19 +28,19 @@ REQUESTS = [(brand, config, r) for brand, config in FW_QUERY_CONFIGS.items() for
 T = TypeVar('T')
 
 
-def chunks(l: List[T], n: int = 128) -> Iterator[List[T]]:
+def chunks(l: list[T], n: int = 128) -> Iterator[list[T]]:
   for i in range(0, len(l), n):
     yield l[i:i + n]
 
 
-def is_brand(brand: str, filter_brand: Optional[str]) -> bool:
+def is_brand(brand: str, filter_brand: str | None) -> bool:
   """Returns if brand matches filter_brand or no brand filter is specified"""
   return filter_brand is None or brand == filter_brand
 
 
-def build_fw_dict(fw_versions: List[capnp.lib.capnp._DynamicStructBuilder],
+def build_fw_dict(fw_versions: list[capnp.lib.capnp._DynamicStructBuilder],
-                  filter_brand: Optional[str] = None) -> Dict[AddrType, Set[bytes]]:
+                  filter_brand: str | None = None) -> dict[AddrType, set[bytes]]:
-  fw_versions_dict: DefaultDict[AddrType, Set[bytes]] = defaultdict(set)
+  fw_versions_dict: defaultdict[AddrType, set[bytes]] = defaultdict(set)
   for fw in fw_versions:
     if is_brand(fw.brand, filter_brand) and not fw.logging:
       sub_addr = fw.subAddress if fw.subAddress != 0 else None
@@ -97,7 +98,7 @@ def match_fw_to_car_fuzzy(live_fw_versions, match_brand=None, log=True, exclude=
     return set()
 
 
-def match_fw_to_car_exact(live_fw_versions, match_brand=None, log=True, extra_fw_versions=None) -> Set[str]:
+def match_fw_to_car_exact(live_fw_versions, match_brand=None, log=True, extra_fw_versions=None) -> set[str]:
   """Do an exact FW match. Returns all cars that match the given
   FW versions for a list of "essential" ECUs. If an ECU is not considered
   essential the FW version can be missing to get a fingerprint, but if it's present it
@@ -164,11 +165,11 @@ def match_fw_to_car(fw_versions, allow_exact=True, allow_fuzzy=True, log=True):
   return True, set()
 
 
-def get_present_ecus(logcan, sendcan, num_pandas=1) -> Set[EcuAddrBusType]:
+def get_present_ecus(logcan, sendcan, num_pandas=1) -> set[EcuAddrBusType]:
   params = Params()
   # queries are split by OBD multiplexing mode
-  queries: Dict[bool, List[List[EcuAddrBusType]]] = {True: [], False: []}
+  queries: dict[bool, list[list[EcuAddrBusType]]] = {True: [], False: []}
-  parallel_queries: Dict[bool, List[EcuAddrBusType]] = {True: [], False: []}
+  parallel_queries: dict[bool, list[EcuAddrBusType]] = {True: [], False: []}
   responses = set()
 
   for brand, config, r in REQUESTS:
@@ -203,7 +204,7 @@ def get_present_ecus(logcan, sendcan, num_pandas=1) -> Set[EcuAddrBusType]:
   return ecu_responses
 
 
-def get_brand_ecu_matches(ecu_rx_addrs: Set[EcuAddrBusType]) -> dict[str, set[AddrType]]:
+def get_brand_ecu_matches(ecu_rx_addrs: set[EcuAddrBusType]) -> dict[str, set[AddrType]]:
   """Returns dictionary of brands and matches with ECUs in their FW versions"""
 
   brand_addrs = {brand: {(addr, subaddr) for _, addr, subaddr in config.get_all_ecus(VERSIONS[brand])} for
@@ -231,7 +232,7 @@ def set_obd_multiplexing(params: Params, obd_multiplexing: bool):
 
 
 def get_fw_versions_ordered(logcan, sendcan, ecu_rx_addrs, timeout=0.1, num_pandas=1, debug=False, progress=False) -> \
-  List[capnp.lib.capnp._DynamicStructBuilder]:
+  list[capnp.lib.capnp._DynamicStructBuilder]:
   """Queries for FW versions ordering brands by likelihood, breaks when exact match is found"""
 
   all_car_fw = []
@@ -254,7 +255,7 @@ def get_fw_versions_ordered(logcan, sendcan, ecu_rx_addrs, timeout=0.1, num_pand
 
 
 def get_fw_versions(logcan, sendcan, query_brand=None, extra=None, timeout=0.1, num_pandas=1, debug=False, progress=False) -> \
-  List[capnp.lib.capnp._DynamicStructBuilder]:
+  list[capnp.lib.capnp._DynamicStructBuilder]:
   versions = VERSIONS.copy()
   params = Params()
 

selfdrive/car/gm/values.py

@@ -1,7 +1,6 @@
 from collections import defaultdict
 from dataclasses import dataclass
 from enum import Enum, StrEnum
-from typing import Dict, List, Union
 
 from cereal import car
 from openpilot.selfdrive.car import dbc_dict
@@ -98,7 +97,7 @@ class GMCarInfo(CarInfo):
       self.footnotes.append(Footnote.OBD_II)
 
 
-CAR_INFO: Dict[str, Union[GMCarInfo, List[GMCarInfo]]] = {
+CAR_INFO: dict[str, GMCarInfo | list[GMCarInfo]] = {
   CAR.HOLDEN_ASTRA: GMCarInfo("Holden Astra 2017"),
   CAR.VOLT: GMCarInfo("Chevrolet Volt 2017-18", min_enable_speed=0, video_link="https://youtu.be/QeMCN_4TFfQ"),
   CAR.CADILLAC_ATS: GMCarInfo("Cadillac ATS Premium Performance 2018"),
@@ -181,7 +180,7 @@ FW_QUERY_CONFIG = FwQueryConfig(
   extra_ecus=[(Ecu.fwdCamera, 0x24b, None)],
 )
 
-DBC: Dict[str, Dict[str, str]] = defaultdict(lambda: dbc_dict('gm_global_a_powertrain_generated', 'gm_global_a_object', chassis_dbc='gm_global_a_chassis'))
+DBC: dict[str, dict[str, str]] = defaultdict(lambda: dbc_dict('gm_global_a_powertrain_generated', 'gm_global_a_object', chassis_dbc='gm_global_a_chassis'))
 
 EV_CAR = {CAR.VOLT, CAR.BOLT_EUV}
 

selfdrive/car/honda/values.py

@@ -1,6 +1,5 @@
 from dataclasses import dataclass
 from enum import Enum, IntFlag, StrEnum
-from typing import Dict, List, Optional, Union
 
 from cereal import car
 from openpilot.common.conversions import Conversions as CV
@@ -116,7 +115,7 @@ class HondaCarInfo(CarInfo):
       self.car_parts = CarParts.common([CarHarness.nidec])
 
 
-CAR_INFO: Dict[str, Optional[Union[HondaCarInfo, List[HondaCarInfo]]]] = {
+CAR_INFO: dict[str, HondaCarInfo | list[HondaCarInfo] | None] = {
   CAR.ACCORD: [
     HondaCarInfo("Honda Accord 2018-22", "All", video_link="https://www.youtube.com/watch?v=mrUwlj3Mi58", min_steer_speed=3. * CV.MPH_TO_MS),
     HondaCarInfo("Honda Inspire 2018", "All", min_steer_speed=3. * CV.MPH_TO_MS),

selfdrive/car/hyundai/values.py

@@ -1,7 +1,6 @@
 import re
 from dataclasses import dataclass
 from enum import Enum, IntFlag, StrEnum
-from typing import Dict, List, Optional, Set, Tuple, Union
 
 from cereal import car
 from panda.python import uds
@@ -157,7 +156,7 @@ class HyundaiCarInfo(CarInfo):
       self.footnotes.insert(0, Footnote.CANFD)
 
 
-CAR_INFO: Dict[str, Optional[Union[HyundaiCarInfo, List[HyundaiCarInfo]]]] = {
+CAR_INFO: dict[str, HyundaiCarInfo | list[HyundaiCarInfo] | None] = {
   CAR.AZERA_6TH_GEN: HyundaiCarInfo("Hyundai Azera 2022", "All", car_parts=CarParts.common([CarHarness.hyundai_k])),
   CAR.AZERA_HEV_6TH_GEN: [
     HyundaiCarInfo("Hyundai Azera Hybrid 2019", "All", car_parts=CarParts.common([CarHarness.hyundai_c])),
@@ -316,7 +315,7 @@ class Buttons:
   CANCEL = 4  # on newer models, this is a pause/resume button
 
 
-def get_platform_codes(fw_versions: List[bytes]) -> Set[Tuple[bytes, Optional[bytes]]]:
+def get_platform_codes(fw_versions: list[bytes]) -> set[tuple[bytes, bytes | None]]:
   # Returns unique, platform-specific identification codes for a set of versions
   codes = set()  # (code-Optional[part], date)
   for fw in fw_versions:
@@ -335,12 +334,12 @@ def get_platform_codes(fw_versions: List[bytes]) -> Set[Tuple[bytes, Optional[by
   return codes
 
 
-def match_fw_to_car_fuzzy(live_fw_versions, offline_fw_versions) -> Set[str]:
+def match_fw_to_car_fuzzy(live_fw_versions, offline_fw_versions) -> set[str]:
   # Non-electric CAN FD platforms often do not have platform code specifiers needed
   # to distinguish between hybrid and ICE. All EVs so far are either exclusively
   # electric or specify electric in the platform code.
   fuzzy_platform_blacklist = {str(c) for c in (CANFD_CAR - EV_CAR - CANFD_FUZZY_WHITELIST)}
-  candidates: Set[str] = set()
+  candidates: set[str] = set()
 
   for candidate, fws in offline_fw_versions.items():
     # Keep track of ECUs which pass all checks (platform codes, within date range)

selfdrive/car/interfaces.py

@@ -4,7 +4,8 @@ import numpy as np
 import tomllib
 from abc import abstractmethod, ABC
 from enum import StrEnum
-from typing import Any, Dict, Optional, Tuple, List, Callable, NamedTuple, cast
+from typing import Any, NamedTuple, cast
+from collections.abc import Callable
 
 from cereal import car
 from openpilot.common.basedir import BASEDIR
@@ -107,7 +108,7 @@ class CarInterfaceBase(ABC):
     return cls.get_params(candidate, gen_empty_fingerprint(), list(), False, False)
 
   @classmethod
-  def get_params(cls, candidate: str, fingerprint: Dict[int, Dict[int, int]], car_fw: List[car.CarParams.CarFw], experimental_long: bool, docs: bool):
+  def get_params(cls, candidate: str, fingerprint: dict[int, dict[int, int]], car_fw: list[car.CarParams.CarFw], experimental_long: bool, docs: bool):
     ret = CarInterfaceBase.get_std_params(candidate)
 
     if hasattr(candidate, "config"):
@@ -131,8 +132,8 @@ class CarInterfaceBase(ABC):
 
   @staticmethod
   @abstractmethod
-  def _get_params(ret: car.CarParams, candidate: str, fingerprint: Dict[int, Dict[int, int]],
+  def _get_params(ret: car.CarParams, candidate: str, fingerprint: dict[int, dict[int, int]],
-                  car_fw: List[car.CarParams.CarFw], experimental_long: bool, docs: bool):
+                  car_fw: list[car.CarParams.CarFw], experimental_long: bool, docs: bool):
     raise NotImplementedError
 
   @staticmethod
@@ -212,7 +213,7 @@ class CarInterfaceBase(ABC):
   def _update(self, c: car.CarControl) -> car.CarState:
     pass
 
-  def update(self, c: car.CarControl, can_strings: List[bytes]) -> car.CarState:
+  def update(self, c: car.CarControl, can_strings: list[bytes]) -> car.CarState:
     # parse can
     for cp in self.can_parsers:
       if cp is not None:
@@ -246,7 +247,7 @@ class CarInterfaceBase(ABC):
     return reader
 
   @abstractmethod
-  def apply(self, c: car.CarControl, now_nanos: int) -> Tuple[car.CarControl.Actuators, List[bytes]]:
+  def apply(self, c: car.CarControl, now_nanos: int) -> tuple[car.CarControl.Actuators, list[bytes]]:
     pass
 
   def create_common_events(self, cs_out, extra_gears=None, pcm_enable=True, allow_enable=True,
@@ -417,11 +418,11 @@ class CarStateBase(ABC):
     return bool(left_blinker_stalk or self.left_blinker_cnt > 0), bool(right_blinker_stalk or self.right_blinker_cnt > 0)
 
   @staticmethod
-  def parse_gear_shifter(gear: Optional[str]) -> car.CarState.GearShifter:
+  def parse_gear_shifter(gear: str | None) -> car.CarState.GearShifter:
     if gear is None:
       return GearShifter.unknown
 
-    d: Dict[str, car.CarState.GearShifter] = {
+    d: dict[str, car.CarState.GearShifter] = {
       'P': GearShifter.park, 'PARK': GearShifter.park,
       'R': GearShifter.reverse, 'REVERSE': GearShifter.reverse,
       'N': GearShifter.neutral, 'NEUTRAL': GearShifter.neutral,
@@ -458,7 +459,7 @@ INTERFACE_ATTR_FILE = {
 
 # interface-specific helpers
 
-def get_interface_attr(attr: str, combine_brands: bool = False, ignore_none: bool = False) -> Dict[str | StrEnum, Any]:
+def get_interface_attr(attr: str, combine_brands: bool = False, ignore_none: bool = False) -> dict[str | StrEnum, Any]:
   # read all the folders in selfdrive/car and return a dict where:
   # - keys are all the car models or brand names
   # - values are attr values from all car folders
@@ -491,7 +492,7 @@ class NanoFFModel:
     self.load_weights(platform)
 
   def load_weights(self, platform: str):
-    with open(self.weights_loc, 'r') as fob:
+    with open(self.weights_loc) as fob:
       self.weights = {k: np.array(v) for k, v in json.load(fob)[platform].items()}
 
   def relu(self, x: np.ndarray):
@@ -506,7 +507,7 @@ class NanoFFModel:
     x = np.dot(x, self.weights['w_4']) + self.weights['b_4']
     return x
 
-  def predict(self, x: List[float], do_sample: bool = False):
+  def predict(self, x: list[float], do_sample: bool = False):
     x = self.forward(np.array(x))
     if do_sample:
       pred = np.random.laplace(x[0], np.exp(x[1]) / self.weights['temperature'])

selfdrive/car/mazda/values.py

@@ -1,6 +1,5 @@
 from dataclasses import dataclass, field
 from enum import StrEnum
-from typing import Dict, List, Union
 
 from cereal import car
 from openpilot.selfdrive.car import dbc_dict
@@ -41,7 +40,7 @@ class MazdaCarInfo(CarInfo):
   car_parts: CarParts = field(default_factory=CarParts.common([CarHarness.mazda]))
 
 
-CAR_INFO: Dict[str, Union[MazdaCarInfo, List[MazdaCarInfo]]] = {
+CAR_INFO: dict[str, MazdaCarInfo | list[MazdaCarInfo]] = {
   CAR.CX5: MazdaCarInfo("Mazda CX-5 2017-21"),
   CAR.CX9: MazdaCarInfo("Mazda CX-9 2016-20"),
   CAR.MAZDA3: MazdaCarInfo("Mazda 3 2017-18"),

selfdrive/car/mock/values.py

@@ -1,5 +1,4 @@
 from enum import StrEnum
-from typing import Dict, List, Optional, Union
 
 from openpilot.selfdrive.car.docs_definitions import CarInfo
 
@@ -8,6 +7,6 @@ class CAR(StrEnum):
   MOCK = 'mock'
 
 
-CAR_INFO: Dict[str, Optional[Union[CarInfo, List[CarInfo]]]] = {
+CAR_INFO: dict[str, CarInfo | list[CarInfo] | None] = {
   CAR.MOCK: None,
 }

selfdrive/car/nissan/values.py

@@ -1,6 +1,5 @@
 from dataclasses import dataclass, field
 from enum import StrEnum
-from typing import Dict, List, Optional, Union
 
 from cereal import car
 from panda.python import uds
@@ -37,7 +36,7 @@ class NissanCarInfo(CarInfo):
   car_parts: CarParts = field(default_factory=CarParts.common([CarHarness.nissan_a]))
 
 
-CAR_INFO: Dict[str, Optional[Union[NissanCarInfo, List[NissanCarInfo]]]] = {
+CAR_INFO: dict[str, NissanCarInfo | list[NissanCarInfo] | None] = {
   CAR.XTRAIL: NissanCarInfo("Nissan X-Trail 2017"),
   CAR.LEAF: NissanCarInfo("Nissan Leaf 2018-23", video_link="https://youtu.be/vaMbtAh_0cY"),
   CAR.LEAF_IC: None,  # same platforms

selfdrive/car/subaru/values.py

@@ -1,6 +1,5 @@
 from dataclasses import dataclass, field
 from enum import Enum, IntFlag
-from typing import List
 
 from cereal import car
 from panda.python import uds
@@ -81,7 +80,7 @@ class Footnote(Enum):
 class SubaruCarInfo(CarInfo):
   package: str = "EyeSight Driver Assistance"
   car_parts: CarParts = field(default_factory=CarParts.common([CarHarness.subaru_a]))
-  footnotes: List[Enum] = field(default_factory=lambda: [Footnote.GLOBAL])
+  footnotes: list[Enum] = field(default_factory=lambda: [Footnote.GLOBAL])
 
   def init_make(self, CP: car.CarParams):
     self.car_parts.parts.extend([Tool.socket_8mm_deep, Tool.pry_tool])

selfdrive/car/tesla/values.py

@@ -1,6 +1,5 @@
 from collections import namedtuple
 from enum import StrEnum
-from typing import Dict, List, Union
 
 from cereal import car
 from openpilot.selfdrive.car import AngleRateLimit, dbc_dict
@@ -17,7 +16,7 @@ class CAR(StrEnum):
   AP2_MODELS = 'TESLA AP2 MODEL S'
 
 
-CAR_INFO: Dict[str, Union[CarInfo, List[CarInfo]]] = {
+CAR_INFO: dict[str, CarInfo | list[CarInfo]] = {
   CAR.AP1_MODELS: CarInfo("Tesla AP1 Model S", "All"),
   CAR.AP2_MODELS: CarInfo("Tesla AP2 Model S", "All"),
 }

selfdrive/car/tests/routes.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-from typing import NamedTuple, Optional
+from typing import NamedTuple
 
 from openpilot.selfdrive.car.chrysler.values import CAR as CHRYSLER
 from openpilot.selfdrive.car.gm.values import CAR as GM
@@ -29,8 +29,8 @@ non_tested_cars = [
 
 class CarTestRoute(NamedTuple):
   route: str
-  car_model: Optional[str]
+  car_model: str | None
-  segment: Optional[int] = None
+  segment: int | None = None
 
 
 routes = [

selfdrive/car/tests/test_docs.py

@@ -20,7 +20,7 @@ class TestCarDocs(unittest.TestCase):
 
   def test_generator(self):
     generated_cars_md = generate_cars_md(self.all_cars, CARS_MD_TEMPLATE)
-    with open(CARS_MD_OUT, "r") as f:
+    with open(CARS_MD_OUT) as f:
       current_cars_md = f.read()
 
     self.assertEqual(generated_cars_md, current_cars_md,
@@ -45,7 +45,7 @@ class TestCarDocs(unittest.TestCase):
     all_car_info_platforms = get_interface_attr("CAR_INFO", combine_brands=True).keys()
     for platform in sorted(interfaces.keys()):
       with self.subTest(platform=platform):
-        self.assertTrue(platform in all_car_info_platforms, "Platform: {} doesn't exist in CarInfo".format(platform))
+        self.assertTrue(platform in all_car_info_platforms, f"Platform: {platform} doesn't exist in CarInfo")
 
   def test_naming_conventions(self):
     # Asserts market-standard car naming conventions by brand

selfdrive/car/tests/test_fingerprints.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 import os
 import sys
-from typing import Dict, List
 
 from openpilot.common.basedir import BASEDIR
 
@@ -64,7 +63,7 @@ def check_can_ignition_conflicts(fingerprints, brands):
 if __name__ == "__main__":
   fingerprints = _get_fingerprints()
 
-  fingerprints_flat: List[Dict] = []
+  fingerprints_flat: list[dict] = []
   car_names = []
   brand_names = []
   for brand in fingerprints:

selfdrive/car/tests/test_lateral_limits.py

@@ -3,7 +3,6 @@ from collections import defaultdict
 import importlib
 from parameterized import parameterized_class
 import sys
-from typing import DefaultDict, Dict
 import unittest
 
 from openpilot.common.realtime import DT_CTRL
@@ -29,7 +28,7 @@ ABOVE_LIMITS_CARS = [
   SUBARU.OUTBACK,
 ]
 
-car_model_jerks: DefaultDict[str, Dict[str, float]] = defaultdict(dict)
+car_model_jerks: defaultdict[str, dict[str, float]] = defaultdict(dict)
 
 
 @parameterized_class('car_model', [(c,) for c in sorted(CAR_MODELS)])

selfdrive/car/tests/test_models.py

@@ -8,7 +8,6 @@ import unittest
 from collections import defaultdict, Counter
 import hypothesis.strategies as st
 from hypothesis import Phase, given, settings
-from typing import List, Optional, Tuple
 from parameterized import parameterized_class
 
 from cereal import messaging, log, car
@@ -40,7 +39,7 @@ MAX_EXAMPLES = int(os.environ.get("MAX_EXAMPLES", "300"))
 CI = os.environ.get("CI", None) is not None
 
 
-def get_test_cases() -> List[Tuple[str, Optional[CarTestRoute]]]:
+def get_test_cases() -> list[tuple[str, CarTestRoute | None]]:
   # build list of test cases
   test_cases = []
   if not len(INTERNAL_SEG_LIST):
@@ -65,14 +64,14 @@ def get_test_cases() -> List[Tuple[str, Optional[CarTestRoute]]]:
 @pytest.mark.slow
 @pytest.mark.shared_download_cache
 class TestCarModelBase(unittest.TestCase):
-  car_model: Optional[str] = None
+  car_model: str | None = None
-  test_route: Optional[CarTestRoute] = None
+  test_route: CarTestRoute | None = None
   test_route_on_bucket: bool = True  # whether the route is on the preserved CI bucket
 
-  can_msgs: List[capnp.lib.capnp._DynamicStructReader]
+  can_msgs: list[capnp.lib.capnp._DynamicStructReader]
   fingerprint: dict[int, dict[int, int]]
-  elm_frame: Optional[int]
+  elm_frame: int | None
-  car_safety_mode_frame: Optional[int]
+  car_safety_mode_frame: int | None
 
   @classmethod
   def get_testing_data_from_logreader(cls, lr):
@@ -408,7 +407,7 @@ class TestCarModelBase(unittest.TestCase):
 
     controls_allowed_prev = False
     CS_prev = car.CarState.new_message()
-    checks = defaultdict(lambda: 0)
+    checks = defaultdict(int)
     controlsd = Controls(CI=self.CI)
     controlsd.initialized = True
     for idx, can in enumerate(self.can_msgs):

selfdrive/car/toyota/values.py

@@ -2,7 +2,6 @@ import re
 from collections import defaultdict
 from dataclasses import dataclass, field
 from enum import Enum, IntFlag, StrEnum
-from typing import Dict, List, Set, Union
 
 from cereal import car
 from openpilot.common.conversions import Conversions as CV
@@ -100,7 +99,7 @@ class ToyotaCarInfo(CarInfo):
   car_parts: CarParts = field(default_factory=CarParts.common([CarHarness.toyota_a]))
 
 
-CAR_INFO: Dict[str, Union[ToyotaCarInfo, List[ToyotaCarInfo]]] = {
+CAR_INFO: dict[str, ToyotaCarInfo | list[ToyotaCarInfo]] = {
   # Toyota
   CAR.ALPHARD_TSS2: [
     ToyotaCarInfo("Toyota Alphard 2019-20"),
@@ -253,7 +252,7 @@ STATIC_DSU_MSGS = [
 ]
 
 
-def get_platform_codes(fw_versions: List[bytes]) -> Dict[bytes, Set[bytes]]:
+def get_platform_codes(fw_versions: list[bytes]) -> dict[bytes, set[bytes]]:
   # Returns sub versions in a dict so comparisons can be made within part-platform-major_version combos
   codes = defaultdict(set)  # Optional[part]-platform-major_version: set of sub_version
   for fw in fw_versions:
@@ -297,7 +296,7 @@ def get_platform_codes(fw_versions: List[bytes]) -> Dict[bytes, Set[bytes]]:
   return dict(codes)
 
 
-def match_fw_to_car_fuzzy(live_fw_versions, offline_fw_versions) -> Set[str]:
+def match_fw_to_car_fuzzy(live_fw_versions, offline_fw_versions) -> set[str]:
   candidates = set()
 
   for candidate, fws in offline_fw_versions.items():

selfdrive/car/volkswagen/values.py

@@ -1,7 +1,6 @@
 from collections import defaultdict, namedtuple
 from dataclasses import dataclass, field
 from enum import Enum, IntFlag, StrEnum
-from typing import Dict, List, Union
 
 from cereal import car
 from panda.python import uds
@@ -151,7 +150,7 @@ class CAR(StrEnum):
 PQ_CARS = {CAR.PASSAT_NMS, CAR.SHARAN_MK2}
 
 
-DBC: Dict[str, Dict[str, str]] = defaultdict(lambda: dbc_dict("vw_mqb_2010", None))
+DBC: dict[str, dict[str, str]] = defaultdict(lambda: dbc_dict("vw_mqb_2010", None))
 for car_type in PQ_CARS:
   DBC[car_type] = dbc_dict("vw_golf_mk4", None)
 
@@ -191,7 +190,7 @@ class VWCarInfo(CarInfo):
       self.car_parts = CarParts([Device.threex_angled_mount, CarHarness.j533])
 
 
-CAR_INFO: Dict[str, Union[VWCarInfo, List[VWCarInfo]]] = {
+CAR_INFO: dict[str, VWCarInfo | list[VWCarInfo]] = {
   CAR.ARTEON_MK1: [
     VWCarInfo("Volkswagen Arteon 2018-23", video_link="https://youtu.be/FAomFKPFlDA"),
     VWCarInfo("Volkswagen Arteon R 2020-23", video_link="https://youtu.be/FAomFKPFlDA"),

selfdrive/controls/lib/alertmanager.py

@@ -3,7 +3,6 @@ import os
 import json
 from collections import defaultdict
 from dataclasses import dataclass
-from typing import List, Dict, Optional
 
 from openpilot.common.basedir import BASEDIR
 from openpilot.common.params import Params
@@ -14,7 +13,7 @@ with open(os.path.join(BASEDIR, "selfdrive/controls/lib/alerts_offroad.json")) a
   OFFROAD_ALERTS = json.load(f)
 
 
-def set_offroad_alert(alert: str, show_alert: bool, extra_text: Optional[str] = None) -> None:
+def set_offroad_alert(alert: str, show_alert: bool, extra_text: str | None = None) -> None:
   if show_alert:
     a = copy.copy(OFFROAD_ALERTS[alert])
     a['extra'] = extra_text or ''
@@ -25,7 +24,7 @@ def set_offroad_alert(alert: str, show_alert: bool, extra_text: Optional[str] =
 
 @dataclass
 class AlertEntry:
-  alert: Optional[Alert] = None
+  alert: Alert | None = None
   start_frame: int = -1
   end_frame: int = -1
 
@@ -34,9 +33,9 @@ class AlertEntry:
 
 class AlertManager:
   def __init__(self):
-    self.alerts: Dict[str, AlertEntry] = defaultdict(AlertEntry)
+    self.alerts: dict[str, AlertEntry] = defaultdict(AlertEntry)
 
-  def add_many(self, frame: int, alerts: List[Alert]) -> None:
+  def add_many(self, frame: int, alerts: list[Alert]) -> None:
     for alert in alerts:
       entry = self.alerts[alert.alert_type]
       entry.alert = alert
@@ -45,7 +44,7 @@ class AlertManager:
       min_end_frame = entry.start_frame + alert.duration
       entry.end_frame = max(frame + 1, min_end_frame)
 
-  def process_alerts(self, frame: int, clear_event_types: set) -> Optional[Alert]:
+  def process_alerts(self, frame: int, clear_event_types: set) -> Alert | None:
     current_alert = AlertEntry()
     for v in self.alerts.values():
       if not v.alert:

selfdrive/controls/lib/events.py

@@ -2,7 +2,7 @@
 import math
 import os
 from enum import IntEnum
-from typing import Dict, Union, Callable, List, Optional
+from collections.abc import Callable
 
 from cereal import log, car
 import cereal.messaging as messaging
@@ -48,12 +48,12 @@ EVENT_NAME = {v: k for k, v in EventName.schema.enumerants.items()}
 
 class Events:
   def __init__(self):
-    self.events: List[int] = []
+    self.events: list[int] = []
-    self.static_events: List[int] = []
+    self.static_events: list[int] = []
     self.events_prev = dict.fromkeys(EVENTS.keys(), 0)
 
   @property
-  def names(self) -> List[int]:
+  def names(self) -> list[int]:
     return self.events
 
   def __len__(self) -> int:
@@ -71,7 +71,7 @@ class Events:
   def contains(self, event_type: str) -> bool:
     return any(event_type in EVENTS.get(e, {}) for e in self.events)
 
-  def create_alerts(self, event_types: List[str], callback_args=None):
+  def create_alerts(self, event_types: list[str], callback_args=None):
     if callback_args is None:
       callback_args = []
 
@@ -132,7 +132,7 @@ class Alert:
     self.creation_delay = creation_delay
 
     self.alert_type = ""
-    self.event_type: Optional[str] = None
+    self.event_type: str | None = None
 
   def __str__(self) -> str:
     return f"{self.alert_text_1}/{self.alert_text_2} {self.priority} {self.visual_alert} {self.audible_alert}"
@@ -333,7 +333,7 @@ def joystick_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster,
 
 
 
-EVENTS: Dict[int, Dict[str, Union[Alert, AlertCallbackType]]] = {
+EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
   # ********** events with no alerts **********
 
   EventName.stockFcw: {},
@@ -965,7 +965,7 @@ if __name__ == '__main__':
   from collections import defaultdict
 
   event_names = {v: k for k, v in EventName.schema.enumerants.items()}
-  alerts_by_type: Dict[str, Dict[Priority, List[str]]] = defaultdict(lambda: defaultdict(list))
+  alerts_by_type: dict[str, dict[Priority, list[str]]] = defaultdict(lambda: defaultdict(list))
 
   CP = car.CarParams.new_message()
   CS = car.CarState.new_message()
@@ -977,7 +977,7 @@ if __name__ == '__main__':
         alert = alert(CP, CS, sm, False, 1)
       alerts_by_type[et][alert.priority].append(event_names[i])
 
-  all_alerts: Dict[str, List[tuple[Priority, List[str]]]] = {}
+  all_alerts: dict[str, list[tuple[Priority, list[str]]]] = {}
   for et, priority_alerts in alerts_by_type.items():
     all_alerts[et] = sorted(priority_alerts.items(), key=lambda x: x[0], reverse=True)
 

selfdrive/controls/lib/vehicle_model.py

@@ -12,7 +12,6 @@ x_dot = A*x + B*u
 
 A depends on longitudinal speed, u [m/s], and vehicle parameters CP
 """
-from typing import Tuple
 
 import numpy as np
 from numpy.linalg import solve
@@ -169,7 +168,7 @@ def kin_ss_sol(sa: float, u: float, VM: VehicleModel) -> np.ndarray:
   return K * sa
 
 
-def create_dyn_state_matrices(u: float, VM: VehicleModel) -> Tuple[np.ndarray, np.ndarray]:
+def create_dyn_state_matrices(u: float, VM: VehicleModel) -> tuple[np.ndarray, np.ndarray]:
   """Returns the A and B matrix for the dynamics system
 
   Args:

selfdrive/controls/radard.py

@@ -2,7 +2,7 @@
 import importlib
 import math
 from collections import deque
-from typing import Optional, Dict, Any
+from typing import Any, Optional
 
 import capnp
 from cereal import messaging, log, car
@@ -125,7 +125,7 @@ def laplacian_pdf(x: float, mu: float, b: float):
   return math.exp(-abs(x-mu)/b)
 
 
-def match_vision_to_track(v_ego: float, lead: capnp._DynamicStructReader, tracks: Dict[int, Track]):
+def match_vision_to_track(v_ego: float, lead: capnp._DynamicStructReader, tracks: dict[int, Track]):
   offset_vision_dist = lead.x[0] - RADAR_TO_CAMERA
 
   def prob(c):
@@ -166,8 +166,8 @@ def get_RadarState_from_vision(lead_msg: capnp._DynamicStructReader, v_ego: floa
   }
 
 
-def get_lead(v_ego: float, ready: bool, tracks: Dict[int, Track], lead_msg: capnp._DynamicStructReader,
+def get_lead(v_ego: float, ready: bool, tracks: dict[int, Track], lead_msg: capnp._DynamicStructReader,
-             model_v_ego: float, low_speed_override: bool = True) -> Dict[str, Any]:
+             model_v_ego: float, low_speed_override: bool = True) -> dict[str, Any]:
   # Determine leads, this is where the essential logic happens
   if len(tracks) > 0 and ready and lead_msg.prob > .5:
     track = match_vision_to_track(v_ego, lead_msg, tracks)
@@ -196,14 +196,14 @@ class RadarD:
   def __init__(self, radar_ts: float, delay: int = 0):
     self.current_time = 0.0
 
-    self.tracks: Dict[int, Track] = {}
+    self.tracks: dict[int, Track] = {}
     self.kalman_params = KalmanParams(radar_ts)
 
     self.v_ego = 0.0
     self.v_ego_hist = deque([0.0], maxlen=delay+1)
     self.last_v_ego_frame = -1
 
-    self.radar_state: Optional[capnp._DynamicStructBuilder] = None
+    self.radar_state: capnp._DynamicStructBuilder | None = None
     self.radar_state_valid = False
 
     self.ready = False

selfdrive/debug/can_print_changes.py

@@ -3,7 +3,6 @@ import argparse
 import binascii
 import time
 from collections import defaultdict
-from typing import Optional
 
 import cereal.messaging as messaging
 from openpilot.selfdrive.debug.can_table import can_table
@@ -96,8 +95,8 @@ if __name__ == "__main__":
 
   args = parser.parse_args()
 
-  init_lr: Optional[LogIterable] = None
+  init_lr: LogIterable | None = None
-  new_lr: Optional[LogIterable] = None
+  new_lr: LogIterable | None = None
 
   if args.init:
     if args.init == '':

selfdrive/debug/check_freq.py

@@ -3,7 +3,7 @@ import argparse
 import numpy as np
 import time
 from collections import defaultdict, deque
-from typing import DefaultDict, Deque, MutableSequence
+from collections.abc import MutableSequence
 
 import cereal.messaging as messaging
 
@@ -19,8 +19,8 @@ if __name__ == "__main__":
   socket_names = args.socket
   sockets = {}
 
-  rcv_times: DefaultDict[str, MutableSequence[float]] = defaultdict(lambda: deque(maxlen=100))
+  rcv_times: defaultdict[str, MutableSequence[float]] = defaultdict(lambda: deque(maxlen=100))
-  valids: DefaultDict[str, Deque[bool]] = defaultdict(lambda: deque(maxlen=100))
+  valids: defaultdict[str, deque[bool]] = defaultdict(lambda: deque(maxlen=100))
 
   t = time.monotonic()
   for name in socket_names:

selfdrive/debug/check_lag.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-from typing import Dict
 
 import cereal.messaging as messaging
 from cereal.services import SERVICE_LIST
@@ -10,7 +9,7 @@ TO_CHECK = ['carState']
 if __name__ == "__main__":
   sm = messaging.SubMaster(TO_CHECK)
 
-  prev_t: Dict[str, float] = {}
+  prev_t: dict[str, float] = {}
 
   while True:
     sm.update()

selfdrive/debug/check_timings.py

@@ -3,13 +3,13 @@
 import sys
 import time
 import numpy as np
-from typing import DefaultDict, MutableSequence
+from collections.abc import MutableSequence
 from collections import defaultdict, deque
 
 import cereal.messaging as messaging
 
 socks = {s: messaging.sub_sock(s, conflate=False) for s in sys.argv[1:]}
-ts: DefaultDict[str, MutableSequence[float]] = defaultdict(lambda: deque(maxlen=100))
+ts: defaultdict[str, MutableSequence[float]] = defaultdict(lambda: deque(maxlen=100))
 
 if __name__ == "__main__":
   while True:

selfdrive/debug/count_events.py

@@ -4,7 +4,7 @@ import math
 import datetime
 from collections import Counter
 from pprint import pprint
-from typing import List, Tuple, cast
+from typing import cast
 
 from cereal.services import SERVICE_LIST
 from openpilot.tools.lib.logreader import LogReader, ReadMode
@@ -15,8 +15,8 @@ if __name__ == "__main__":
   cams = [s for s in SERVICE_LIST if s.endswith('CameraState')]
   cnt_cameras = dict.fromkeys(cams, 0)
 
-  events: List[Tuple[float, set[str]]] = []
+  events: list[tuple[float, set[str]]] = []
-  alerts: List[Tuple[float, str]] = []
+  alerts: list[tuple[float, str]] = []
   start_time = math.inf
   end_time = -math.inf
   ignition_off = None

selfdrive/debug/cpu_usage_stat.py

@@ -66,12 +66,12 @@ if __name__ == "__main__":
   for p in psutil.process_iter():
     if p == psutil.Process():
       continue
-    matched = any(l for l in p.cmdline() if any(pn for pn in monitored_proc_names if re.match(r'.*{}.*'.format(pn), l, re.M | re.I)))
+    matched = any(l for l in p.cmdline() if any(pn for pn in monitored_proc_names if re.match(fr'.*{pn}.*', l, re.M | re.I)))
     if matched:
       k = ' '.join(p.cmdline())
       print('Add monitored proc:', k)
       stats[k] = {'cpu_samples': defaultdict(list), 'min': defaultdict(lambda: None), 'max': defaultdict(lambda: None),
-                  'avg': defaultdict(lambda: 0.0), 'last_cpu_times': None, 'last_sys_time': None}
+                  'avg': defaultdict(float), 'last_cpu_times': None, 'last_sys_time': None}
       stats[k]['last_sys_time'] = timer()
       stats[k]['last_cpu_times'] = p.cpu_times()
       monitored_procs.append(p)

selfdrive/debug/format_fingerprints.py

@@ -67,7 +67,7 @@ def format_brand_fw_versions(brand, extra_fw_versions: None | dict[str, dict[tup
   extra_fw_versions = extra_fw_versions or {}
 
   fingerprints_file = os.path.join(BASEDIR, f"selfdrive/car/{brand}/fingerprints.py")
-  with open(fingerprints_file, "r") as f:
+  with open(fingerprints_file) as f:
     comments = [line for line in f.readlines() if line.startswith("#") and "noqa" not in line]
 
   with open(fingerprints_file, "w") as f:

selfdrive/debug/internal/measure_modeld_packet_drop.py

@@ -1,12 +1,11 @@
 #!/usr/bin/env python3
 import cereal.messaging as messaging
-from typing import Optional
 
 if __name__ == "__main__":
   modeld_sock = messaging.sub_sock("modelV2")
 
   last_frame_id = None
-  start_t: Optional[int] = None
+  start_t: int | None = None
   frame_cnt = 0
   dropped = 0
 

selfdrive/debug/live_cpu_and_temp.py

@@ -5,7 +5,6 @@ from collections import defaultdict
 
 from cereal.messaging import SubMaster
 from openpilot.common.numpy_fast import mean
-from typing import Optional, Dict
 
 def cputime_total(ct):
   return ct.user + ct.nice + ct.system + ct.idle + ct.iowait + ct.irq + ct.softirq
@@ -42,8 +41,8 @@ if __name__ == "__main__":
   total_times = [0.]*8
   busy_times = [0.]*8
 
-  prev_proclog: Optional[capnp._DynamicStructReader] = None
+  prev_proclog: capnp._DynamicStructReader | None = None
-  prev_proclog_t: Optional[int] = None
+  prev_proclog_t: int | None = None
 
   while True:
     sm.update()
@@ -76,7 +75,7 @@ if __name__ == "__main__":
       print(f"CPU {100.0 * mean(cores):.2f}% - RAM: {last_mem:.2f}% - Temp {last_temp:.2f}C")
 
       if args.cpu and prev_proclog is not None and prev_proclog_t is not None:
-        procs: Dict[str, float] = defaultdict(float)
+        procs: dict[str, float] = defaultdict(float)
         dt = (sm.logMonoTime['procLog'] - prev_proclog_t) / 1e9
         for proc in m.procs:
           try:

selfdrive/locationd/calibrationd.py

@@ -10,7 +10,7 @@ import gc
 import os
 import capnp
 import numpy as np
-from typing import List, NoReturn, Optional
+from typing import NoReturn
 
 from cereal import log
 import cereal.messaging as messaging
@@ -89,7 +89,7 @@ class Calibrator:
                   valid_blocks: int = 0,
                   wide_from_device_euler_init: np.ndarray = WIDE_FROM_DEVICE_EULER_INIT,
                   height_init: np.ndarray = HEIGHT_INIT,
-                  smooth_from: Optional[np.ndarray] = None) -> None:
+                  smooth_from: np.ndarray | None = None) -> None:
     if not np.isfinite(rpy_init).all():
       self.rpy = RPY_INIT.copy()
     else:
@@ -125,7 +125,7 @@ class Calibrator:
       self.old_rpy = smooth_from
       self.old_rpy_weight = 1.0
 
-  def get_valid_idxs(self) -> List[int]:
+  def get_valid_idxs(self) -> list[int]:
     # exclude current block_idx from validity window
     before_current = list(range(self.block_idx))
     after_current = list(range(min(self.valid_blocks, self.block_idx + 1), self.valid_blocks))
@@ -175,12 +175,12 @@ class Calibrator:
     else:
       return self.rpy
 
-  def handle_cam_odom(self, trans: List[float],
+  def handle_cam_odom(self, trans: list[float],
-                            rot: List[float],
+                            rot: list[float],
-                            wide_from_device_euler: List[float],
+                            wide_from_device_euler: list[float],
-                            trans_std: List[float],
+                            trans_std: list[float],
-                            road_transform_trans: List[float],
+                            road_transform_trans: list[float],
-                            road_transform_trans_std: List[float]) -> Optional[np.ndarray]:
+                            road_transform_trans_std: list[float]) -> np.ndarray | None:
     self.old_rpy_weight = max(0.0, self.old_rpy_weight - 1/SMOOTH_CYCLES)
 
     straight_and_fast = ((self.v_ego > MIN_SPEED_FILTER) and (trans[0] > MIN_SPEED_FILTER) and (abs(rot[2]) < MAX_YAW_RATE_FILTER))

selfdrive/locationd/helpers.py

@@ -1,5 +1,5 @@
 import numpy as np
-from typing import List, Optional, Tuple, Any
+from typing import Any
 
 from cereal import log
 
@@ -12,7 +12,7 @@ class NPQueue:
   def __len__(self) -> int:
     return len(self.arr)
 
-  def append(self, pt: List[float]) -> None:
+  def append(self, pt: list[float]) -> None:
     if len(self.arr) < self.maxlen:
       self.arr = np.append(self.arr, [pt], axis=0)
     else:
@@ -21,7 +21,7 @@ class NPQueue:
 
 
 class PointBuckets:
-  def __init__(self, x_bounds: List[Tuple[float, float]], min_points: List[float], min_points_total: int, points_per_bucket: int, rowsize: int) -> None:
+  def __init__(self, x_bounds: list[tuple[float, float]], min_points: list[float], min_points_total: int, points_per_bucket: int, rowsize: int) -> None:
     self.x_bounds = x_bounds
     self.buckets = {bounds: NPQueue(maxlen=points_per_bucket, rowsize=rowsize) for bounds in x_bounds}
     self.buckets_min_points = dict(zip(x_bounds, min_points, strict=True))
@@ -41,13 +41,13 @@ class PointBuckets:
   def add_point(self, x: float, y: float, bucket_val: float) -> None:
     raise NotImplementedError
 
-  def get_points(self, num_points: Optional[int] = None) -> Any:
+  def get_points(self, num_points: int | None = None) -> Any:
     points = np.vstack([x.arr for x in self.buckets.values()])
     if num_points is None:
       return points
     return points[np.random.choice(np.arange(len(points)), min(len(points), num_points), replace=False)]
 
-  def load_points(self, points: List[List[float]]) -> None:
+  def load_points(self, points: list[list[float]]) -> None:
     for point in points:
       self.add_point(*point)
 

selfdrive/locationd/models/car_kf.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 import math
 import sys
-from typing import Any, Dict
+from typing import Any
 
 import numpy as np
 
@@ -70,7 +70,7 @@ class CarKalman(KalmanFilter):
   ])
   P_initial = Q.copy()
 
-  obs_noise: Dict[int, Any] = {
+  obs_noise: dict[int, Any] = {
     ObservationKind.STEER_ANGLE: np.atleast_2d(math.radians(0.05)**2),
     ObservationKind.ANGLE_OFFSET_FAST: np.atleast_2d(math.radians(10.0)**2),
     ObservationKind.ROAD_ROLL: np.atleast_2d(math.radians(1.0)**2),

selfdrive/manager/build.py

@@ -2,7 +2,6 @@
 import os
 import subprocess
 from pathlib import Path
-from typing import List
 
 # NOTE: Do NOT import anything here that needs be built (e.g. params)
 from openpilot.common.basedir import BASEDIR
@@ -29,7 +28,7 @@ def build(spinner: Spinner, dirty: bool = False, minimal: bool = False) -> None:
 
   # building with all cores can result in using too
   # much memory, so retry with less parallelism
-  compile_output: List[bytes] = []
+  compile_output: list[bytes] = []
   for n in (nproc, nproc/2, 1):
     compile_output.clear()
     scons: subprocess.Popen = subprocess.Popen(["scons", f"-j{int(n)}", "--cache-populate", *extra_args], cwd=BASEDIR, env=env, stderr=subprocess.PIPE)

selfdrive/manager/manager.py

@@ -4,7 +4,6 @@ import os
 import signal
 import sys
 import traceback
-from typing import List, Tuple, Union
 
 from cereal import log
 import cereal.messaging as messaging
@@ -33,7 +32,7 @@ def manager_init() -> None:
   if is_release_branch():
     params.clear_all(ParamKeyType.DEVELOPMENT_ONLY)
 
-  default_params: List[Tuple[str, Union[str, bytes]]] = [
+  default_params: list[tuple[str, str | bytes]] = [
     ("CompletedTrainingVersion", "0"),
     ("DisengageOnAccelerator", "0"),
     ("GsmMetered", "1"),
@@ -121,7 +120,7 @@ def manager_thread() -> None:
 
   params = Params()
 
-  ignore: List[str] = []
+  ignore: list[str] = []
   if params.get("DongleId", encoding='utf8') in (None, UNREGISTERED_DONGLE_ID):
     ignore += ["manage_athenad", "uploader"]
   if os.getenv("NOBOARD") is not None:
@@ -154,7 +153,7 @@ def manager_thread() -> None:
 
     ensure_running(managed_processes.values(), started, params=params, CP=sm['carParams'], not_run=ignore)
 
-    running = ' '.join("%s%s\u001b[0m" % ("\u001b[32m" if p.proc.is_alive() else "\u001b[31m", p.name)
+    running = ' '.join("{}{}\u001b[0m".format("\u001b[32m" if p.proc.is_alive() else "\u001b[31m", p.name)
                        for p in managed_processes.values() if p.proc)
     print(running)
     cloudlog.debug(running)

selfdrive/manager/process.py

@@ -4,7 +4,7 @@ import signal
 import struct
 import time
 import subprocess
-from typing import Optional, Callable, List, ValuesView
+from collections.abc import Callable, ValuesView
 from abc import ABC, abstractmethod
 from multiprocessing import Process
 
@@ -47,7 +47,7 @@ def launcher(proc: str, name: str) -> None:
     raise
 
 
-def nativelauncher(pargs: List[str], cwd: str, name: str) -> None:
+def nativelauncher(pargs: list[str], cwd: str, name: str) -> None:
   os.environ['MANAGER_DAEMON'] = name
 
   # exec the process
@@ -67,12 +67,12 @@ class ManagerProcess(ABC):
   daemon = False
   sigkill = False
   should_run: Callable[[bool, Params, car.CarParams], bool]
-  proc: Optional[Process] = None
+  proc: Process | None = None
   enabled = True
   name = ""
 
   last_watchdog_time = 0
-  watchdog_max_dt: Optional[int] = None
+  watchdog_max_dt: int | None = None
   watchdog_seen = False
   shutting_down = False
 
@@ -109,7 +109,7 @@ class ManagerProcess(ABC):
     else:
       self.watchdog_seen = True
 
-  def stop(self, retry: bool = True, block: bool = True, sig: Optional[signal.Signals] = None) -> Optional[int]:
+  def stop(self, retry: bool = True, block: bool = True, sig: signal.Signals | None = None) -> int | None:
     if self.proc is None:
       return None
 
@@ -274,7 +274,7 @@ class DaemonProcess(ManagerProcess):
 
 
 def ensure_running(procs: ValuesView[ManagerProcess], started: bool, params=None, CP: car.CarParams=None,
-                   not_run: Optional[List[str]]=None) -> List[ManagerProcess]:
+                   not_run: list[str] | None=None) -> list[ManagerProcess]:
   if not_run is None:
     not_run = []
 

selfdrive/modeld/dmonitoringmodeld.py

@@ -6,7 +6,6 @@ import time
 import ctypes
 import numpy as np
 from pathlib import Path
-from typing import Tuple, Dict
 
 from cereal import messaging
 from cereal.messaging import PubMaster, SubMaster
@@ -53,7 +52,7 @@ class DMonitoringModelResult(ctypes.Structure):
     ("wheel_on_right_prob", ctypes.c_float)]
 
 class ModelState:
-  inputs: Dict[str, np.ndarray]
+  inputs: dict[str, np.ndarray]
   output: np.ndarray
   model: ModelRunner
 
@@ -68,7 +67,7 @@ class ModelState:
     self.model.addInput("input_img", None)
     self.model.addInput("calib", self.inputs['calib'])
 
-  def run(self, buf:VisionBuf, calib:np.ndarray) -> Tuple[np.ndarray, float]:
+  def run(self, buf:VisionBuf, calib:np.ndarray) -> tuple[np.ndarray, float]:
     self.inputs['calib'][:] = calib
 
     v_offset = buf.height - MODEL_HEIGHT

selfdrive/modeld/fill_model_msg.py

@@ -1,7 +1,6 @@
 import os
 import capnp
 import numpy as np
-from typing import Dict
 from cereal import log
 from openpilot.selfdrive.modeld.constants import ModelConstants, Plan, Meta
 
@@ -42,7 +41,7 @@ def fill_xyvat(builder, t, x, y, v, a, x_std=None, y_std=None, v_std=None, a_std
   if a_std is not None:
     builder.aStd = a_std.tolist()
 
-def fill_model_msg(msg: capnp._DynamicStructBuilder, net_output_data: Dict[str, np.ndarray], publish_state: PublishState,
+def fill_model_msg(msg: capnp._DynamicStructBuilder, net_output_data: dict[str, np.ndarray], publish_state: PublishState,
                    vipc_frame_id: int, vipc_frame_id_extra: int, frame_id: int, frame_drop: float,
                    timestamp_eof: int, timestamp_llk: int, model_execution_time: float,
                    nav_enabled: bool, valid: bool) -> None:
@@ -174,7 +173,7 @@ def fill_model_msg(msg: capnp._DynamicStructBuilder, net_output_data: Dict[str,
   if SEND_RAW_PRED:
     modelV2.rawPredictions = net_output_data['raw_pred'].tobytes()
 
-def fill_pose_msg(msg: capnp._DynamicStructBuilder, net_output_data: Dict[str, np.ndarray],
+def fill_pose_msg(msg: capnp._DynamicStructBuilder, net_output_data: dict[str, np.ndarray],
                   vipc_frame_id: int, vipc_dropped_frames: int, timestamp_eof: int, live_calib_seen: bool) -> None:
   msg.valid = live_calib_seen & (vipc_dropped_frames < 1)
   cameraOdometry = msg.cameraOdometry

selfdrive/modeld/get_model_metadata.py

@@ -4,9 +4,8 @@ import pathlib
 import onnx
 import codecs
 import pickle
-from typing import Tuple
 
-def get_name_and_shape(value_info:onnx.ValueInfoProto) -> Tuple[str, Tuple[int,...]]:
+def get_name_and_shape(value_info:onnx.ValueInfoProto) -> tuple[str, tuple[int,...]]:
   shape = tuple([int(dim.dim_value) for dim in value_info.type.tensor_type.shape.dim])
   name = value_info.name
   return name, shape

selfdrive/modeld/modeld.py

@@ -6,7 +6,6 @@ import numpy as np
 import cereal.messaging as messaging
 from cereal import car, log
 from pathlib import Path
-from typing import Dict, Optional
 from setproctitle import setproctitle
 from cereal.messaging import PubMaster, SubMaster
 from cereal.visionipc import VisionIpcClient, VisionStreamType, VisionBuf
@@ -45,7 +44,7 @@ class FrameMeta:
 class ModelState:
   frame: ModelFrame
   wide_frame: ModelFrame
-  inputs: Dict[str, np.ndarray]
+  inputs: dict[str, np.ndarray]
   output: np.ndarray
   prev_desire: np.ndarray  # for tracking the rising edge of the pulse
   model: ModelRunner
@@ -78,14 +77,14 @@ class ModelState:
     for k,v in self.inputs.items():
       self.model.addInput(k, v)
 
-  def slice_outputs(self, model_outputs: np.ndarray) -> Dict[str, np.ndarray]:
+  def slice_outputs(self, model_outputs: np.ndarray) -> dict[str, np.ndarray]:
     parsed_model_outputs = {k: model_outputs[np.newaxis, v] for k,v in self.output_slices.items()}
     if SEND_RAW_PRED:
       parsed_model_outputs['raw_pred'] = model_outputs.copy()
     return parsed_model_outputs
 
   def run(self, buf: VisionBuf, wbuf: VisionBuf, transform: np.ndarray, transform_wide: np.ndarray,
-                inputs: Dict[str, np.ndarray], prepare_only: bool) -> Optional[Dict[str, np.ndarray]]:
+                inputs: dict[str, np.ndarray], prepare_only: bool) -> dict[str, np.ndarray] | None:
     # Model decides when action is completed, so desire input is just a pulse triggered on rising edge
     inputs['desire'][0] = 0
     self.inputs['desire'][:-ModelConstants.DESIRE_LEN] = self.inputs['desire'][ModelConstants.DESIRE_LEN:]
@@ -276,7 +275,7 @@ def main(demo=False):
     if prepare_only:
       cloudlog.error(f"skipping model eval. Dropped {vipc_dropped_frames} frames")
 
-    inputs:Dict[str, np.ndarray] = {
+    inputs:dict[str, np.ndarray] = {
       'desire': vec_desire,
       'traffic_convention': traffic_convention,
       'lateral_control_params': lateral_control_params,

selfdrive/modeld/navmodeld.py

@@ -5,7 +5,6 @@ import time
 import ctypes
 import numpy as np
 from pathlib import Path
-from typing import Tuple, Dict
 
 from cereal import messaging
 from cereal.messaging import PubMaster, SubMaster
@@ -41,7 +40,7 @@ class NavModelResult(ctypes.Structure):
     ("features", ctypes.c_float*NAV_FEATURE_LEN)]
 
 class ModelState:
-  inputs: Dict[str, np.ndarray]
+  inputs: dict[str, np.ndarray]
   output: np.ndarray
   model: ModelRunner
 
@@ -52,7 +51,7 @@ class ModelState:
     self.model = ModelRunner(MODEL_PATHS, self.output, Runtime.DSP, True, None)
     self.model.addInput("input_img", None)
 
-  def run(self, buf:np.ndarray) -> Tuple[np.ndarray, float]:
+  def run(self, buf:np.ndarray) -> tuple[np.ndarray, float]:
     self.inputs['input_img'][:] = buf
 
     t1 = time.perf_counter()

selfdrive/modeld/parse_model_outputs.py

@@ -1,5 +1,4 @@
 import numpy as np
-from typing import Dict
 from openpilot.selfdrive.modeld.constants import ModelConstants
 
 def sigmoid(x):
@@ -82,7 +81,7 @@ class Parser:
     outs[name] = pred_mu_final.reshape(final_shape)
     outs[name + '_stds'] = pred_std_final.reshape(final_shape)
 
-  def parse_outputs(self, outs: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]:
+  def parse_outputs(self, outs: dict[str, np.ndarray]) -> dict[str, np.ndarray]:
     self.parse_mdn('plan', outs, in_N=ModelConstants.PLAN_MHP_N, out_N=ModelConstants.PLAN_MHP_SELECTION,
                    out_shape=(ModelConstants.IDX_N,ModelConstants.PLAN_WIDTH))
     self.parse_mdn('lane_lines', outs, in_N=0, out_N=0, out_shape=(ModelConstants.NUM_LANE_LINES,ModelConstants.IDX_N,ModelConstants.LANE_LINES_WIDTH))

selfdrive/modeld/runners/onnxmodel.py

@@ -3,7 +3,7 @@ import itertools
 import os
 import sys
 import numpy as np
-from typing import Tuple, Dict, Union, Any
+from typing import Any
 
 from openpilot.selfdrive.modeld.runners.runmodel_pyx import RunModel
 
@@ -38,7 +38,7 @@ def create_ort_session(path, fp16_to_fp32):
   options = ort.SessionOptions()
   options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_DISABLE_ALL
 
-  provider: Union[str, Tuple[str, Dict[Any, Any]]]
+  provider: str | tuple[str, dict[Any, Any]]
   if 'OpenVINOExecutionProvider' in ort.get_available_providers() and 'ONNXCPU' not in os.environ:
     provider = 'OpenVINOExecutionProvider'
   elif 'CUDAExecutionProvider' in ort.get_available_providers() and 'ONNXCPU' not in os.environ:

selfdrive/navd/helpers.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import json
 import math
-from typing import Any, Dict, List, Optional, Tuple, Union, cast
+from typing import Any, cast
 
 from openpilot.common.conversions import Conversions
 from openpilot.common.numpy_fast import clip
@@ -22,13 +22,13 @@ class Coordinate:
   def __init__(self, latitude: float, longitude: float) -> None:
     self.latitude = latitude
     self.longitude = longitude
-    self.annotations: Dict[str, float] = {}
+    self.annotations: dict[str, float] = {}
 
   @classmethod
-  def from_mapbox_tuple(cls, t: Tuple[float, float]) -> Coordinate:
+  def from_mapbox_tuple(cls, t: tuple[float, float]) -> Coordinate:
     return cls(t[1], t[0])
 
-  def as_dict(self) -> Dict[str, float]:
+  def as_dict(self) -> dict[str, float]:
     return {'latitude': self.latitude, 'longitude': self.longitude}
 
   def __str__(self) -> str:
@@ -83,7 +83,7 @@ def minimum_distance(a: Coordinate, b: Coordinate, p: Coordinate):
   return projection.distance_to(p)
 
 
-def distance_along_geometry(geometry: List[Coordinate], pos: Coordinate) -> float:
+def distance_along_geometry(geometry: list[Coordinate], pos: Coordinate) -> float:
   if len(geometry) <= 2:
     return geometry[0].distance_to(pos)
 
@@ -106,7 +106,7 @@ def distance_along_geometry(geometry: List[Coordinate], pos: Coordinate) -> floa
   return total_distance_closest
 
 
-def coordinate_from_param(param: str, params: Optional[Params] = None) -> Optional[Coordinate]:
+def coordinate_from_param(param: str, params: Params | None = None) -> Coordinate | None:
   if params is None:
     params = Params()
 
@@ -130,7 +130,7 @@ def string_to_direction(direction: str) -> str:
   return 'none'
 
 
-def maxspeed_to_ms(maxspeed: Dict[str, Union[str, float]]) -> float:
+def maxspeed_to_ms(maxspeed: dict[str, str | float]) -> float:
   unit = cast(str, maxspeed['unit'])
   speed = cast(float, maxspeed['speed'])
   return SPEED_CONVERSIONS[unit] * speed
@@ -140,7 +140,7 @@ def field_valid(dat: dict, field: str) -> bool:
   return field in dat and dat[field] is not None
 
 
-def parse_banner_instructions(banners: Any, distance_to_maneuver: float = 0.0) -> Optional[Dict[str, Any]]:
+def parse_banner_instructions(banners: Any, distance_to_maneuver: float = 0.0) -> dict[str, Any] | None:
   if not len(banners):
     return None
 

selfdrive/statsd.py

@@ -5,7 +5,7 @@ import time
 from pathlib import Path
 from collections import defaultdict
 from datetime import datetime, timezone
-from typing import NoReturn, Union, List, Dict
+from typing import NoReturn
 
 from openpilot.common.params import Params
 from cereal.messaging import SubMaster
@@ -61,7 +61,7 @@ class StatLog:
 
 def main() -> NoReturn:
   dongle_id = Params().get("DongleId", encoding='utf-8')
-  def get_influxdb_line(measurement: str, value: Union[float, Dict[str, float]],  timestamp: datetime, tags: dict) -> str:
+  def get_influxdb_line(measurement: str, value: float | dict[str, float],  timestamp: datetime, tags: dict) -> str:
     res = f"{measurement}"
     for k, v in tags.items():
       res += f",{k}={str(v)}"
@@ -102,7 +102,7 @@ def main() -> NoReturn:
   idx = 0
   last_flush_time = time.monotonic()
   gauges = {}
-  samples: Dict[str, List[float]] = defaultdict(list)
+  samples: dict[str, list[float]] = defaultdict(list)
   try:
     while True:
       started_prev = sm['deviceState'].started

selfdrive/test/ciui.py

@@ -11,7 +11,7 @@ from openpilot.selfdrive.ui.qt.python_helpers import set_main_window
 
 class Window(QWidget):
   def __init__(self, parent=None):
-    super(Window, self).__init__(parent)
+    super().__init__(parent)
 
     layout = QVBoxLayout()
     self.setLayout(layout)
@@ -47,7 +47,7 @@ class Window(QWidget):
 
   def update(self):
     for cmd, label in self.labels.items():
-      out = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+      out = subprocess.run(cmd, capture_output=True,
                            shell=True, check=False, encoding='utf8').stdout
       label.setText(out.strip())
 

selfdrive/test/fuzzy_generation.py

@@ -1,6 +1,7 @@
 import capnp
 import hypothesis.strategies as st
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any
+from collections.abc import Callable
 
 from cereal import log
 
@@ -12,7 +13,7 @@ class FuzzyGenerator:
     self.draw = draw
     self.real_floats = real_floats
 
-  def generate_native_type(self, field: str) -> st.SearchStrategy[Union[bool, int, float, str, bytes]]:
+  def generate_native_type(self, field: str) -> st.SearchStrategy[bool | int | float | str | bytes]:
     def floats(**kwargs) -> st.SearchStrategy[float]:
       allow_nan = not self.real_floats
       allow_infinity = not self.real_floats
@@ -67,18 +68,18 @@ class FuzzyGenerator:
     else:
       return self.generate_struct(field.schema)
 
-  def generate_struct(self, schema: capnp.lib.capnp._StructSchema, event: Optional[str] = None) -> st.SearchStrategy[Dict[str, Any]]:
+  def generate_struct(self, schema: capnp.lib.capnp._StructSchema, event: str | None = None) -> st.SearchStrategy[dict[str, Any]]:
-    full_fill: List[str] = list(schema.non_union_fields)
+    full_fill: list[str] = list(schema.non_union_fields)
-    single_fill: List[str] = [event] if event else [self.draw(st.sampled_from(schema.union_fields))] if schema.union_fields else []
+    single_fill: list[str] = [event] if event else [self.draw(st.sampled_from(schema.union_fields))] if schema.union_fields else []
     return st.fixed_dictionaries({field: self.generate_field(schema.fields[field]) for field in full_fill + single_fill})
 
   @classmethod
-  def get_random_msg(cls, draw: DrawType, struct: capnp.lib.capnp._StructModule, real_floats: bool = False) -> Dict[str, Any]:
+  def get_random_msg(cls, draw: DrawType, struct: capnp.lib.capnp._StructModule, real_floats: bool = False) -> dict[str, Any]:
     fg = cls(draw, real_floats=real_floats)
-    data: Dict[str, Any] = draw(fg.generate_struct(struct.schema))
+    data: dict[str, Any] = draw(fg.generate_struct(struct.schema))
     return data
 
   @classmethod
-  def get_random_event_msg(cls, draw: DrawType, events: List[str], real_floats: bool = False) -> List[Dict[str, Any]]:
+  def get_random_event_msg(cls, draw: DrawType, events: list[str], real_floats: bool = False) -> list[dict[str, Any]]:
     fg = cls(draw, real_floats=real_floats)
     return [draw(fg.generate_struct(log.Event.schema, e)) for e in sorted(events)]

selfdrive/test/helpers.py

@@ -72,7 +72,7 @@ def noop(*args, **kwargs):
 
 
 def read_segment_list(segment_list_path):
-  with open(segment_list_path, "r") as f:
+  with open(segment_list_path) as f:
     seg_list = f.read().splitlines()
 
   return [(platform[2:], segment) for platform, segment in zip(seg_list[::2], seg_list[1::2], strict=True)]

selfdrive/test/process_replay/capture.py

@@ -1,7 +1,7 @@
 import os
 import sys
 
-from typing import Tuple, no_type_check
+from typing import no_type_check
 
 class FdRedirect:
   def __init__(self, file_prefix: str, fd: int):
@@ -53,7 +53,7 @@ class ProcessOutputCapture:
     self.stdout_redirect.link()
     self.stderr_redirect.link()
 
-  def read_outerr(self) -> Tuple[str, str]:
+  def read_outerr(self) -> tuple[str, str]:
     out_str = self.stdout_redirect.read().decode()
     err_str = self.stderr_redirect.read().decode()
     return out_str, err_str

selfdrive/test/process_replay/compare_logs.py

@@ -5,7 +5,6 @@ import capnp
 import numbers
 import dictdiffer
 from collections import Counter
-from typing import Dict
 
 from openpilot.tools.lib.logreader import LogReader
 
@@ -97,7 +96,7 @@ def format_process_diff(diff):
     diff_short += f"        {diff}\n"
     diff_long += f"\t{diff}\n"
   else:
-    cnt: Dict[str, int] = {}
+    cnt: dict[str, int] = {}
     for d in diff:
       diff_long += f"\t{str(d)}\n"
 

selfdrive/test/process_replay/process_replay.py

@@ -8,7 +8,8 @@ import signal
 import platform
 from collections import OrderedDict
 from dataclasses import dataclass, field
-from typing import Dict, List, Optional, Callable, Union, Any, Iterable, Tuple
+from typing import Any
+from collections.abc import Callable, Iterable
 from tqdm import tqdm
 import capnp
 
@@ -36,9 +37,9 @@ FAKEDATA = os.path.join(PROC_REPLAY_DIR, "fakedata/")
 
 class DummySocket:
   def __init__(self):
-    self.data: List[bytes] = []
+    self.data: list[bytes] = []
 
-  def receive(self, non_blocking: bool = False) -> Optional[bytes]:
+  def receive(self, non_blocking: bool = False) -> bytes | None:
     if non_blocking:
       return None
 
@@ -128,21 +129,21 @@ class ReplayContext:
 @dataclass
 class ProcessConfig:
   proc_name: str
-  pubs: List[str]
+  pubs: list[str]
-  subs: List[str]
+  subs: list[str]
-  ignore: List[str]
+  ignore: list[str]
-  config_callback: Optional[Callable] = None
+  config_callback: Callable | None = None
-  init_callback: Optional[Callable] = None
+  init_callback: Callable | None = None
-  should_recv_callback: Optional[Callable] = None
+  should_recv_callback: Callable | None = None
-  tolerance: Optional[float] = None
+  tolerance: float | None = None
   processing_time: float = 0.001
   timeout: int = 30
   simulation: bool = True
-  main_pub: Optional[str] = None
+  main_pub: str | None = None
   main_pub_drained: bool = True
-  vision_pubs: List[str] = field(default_factory=list)
+  vision_pubs: list[str] = field(default_factory=list)
-  ignore_alive_pubs: List[str] = field(default_factory=list)
+  ignore_alive_pubs: list[str] = field(default_factory=list)
-  unlocked_pubs: List[str] = field(default_factory=list)
+  unlocked_pubs: list[str] = field(default_factory=list)
 
 
 class ProcessContainer:
@@ -150,25 +151,25 @@ class ProcessContainer:
     self.prefix = OpenpilotPrefix(clean_dirs_on_exit=False)
     self.cfg = copy.deepcopy(cfg)
     self.process = copy.deepcopy(managed_processes[cfg.proc_name])
-    self.msg_queue: List[capnp._DynamicStructReader] = []
+    self.msg_queue: list[capnp._DynamicStructReader] = []
     self.cnt = 0
-    self.pm: Optional[messaging.PubMaster] = None
+    self.pm: messaging.PubMaster | None = None
-    self.sockets: Optional[List[messaging.SubSocket]] = None
+    self.sockets: list[messaging.SubSocket] | None = None
-    self.rc: Optional[ReplayContext] = None
+    self.rc: ReplayContext | None = None
-    self.vipc_server: Optional[VisionIpcServer] = None
+    self.vipc_server: VisionIpcServer | None = None
-    self.environ_config: Optional[Dict[str, Any]] = None
+    self.environ_config: dict[str, Any] | None = None
-    self.capture: Optional[ProcessOutputCapture] = None
+    self.capture: ProcessOutputCapture | None = None
 
   @property
   def has_empty_queue(self) -> bool:
     return len(self.msg_queue) == 0
 
   @property
-  def pubs(self) -> List[str]:
+  def pubs(self) -> list[str]:
     return self.cfg.pubs
 
   @property
-  def subs(self) -> List[str]:
+  def subs(self) -> list[str]:
     return self.cfg.subs
 
   def _clean_env(self):
@@ -180,7 +181,7 @@ class ProcessContainer:
       if k in os.environ:
         del os.environ[k]
 
-  def _setup_env(self, params_config: Dict[str, Any], environ_config: Dict[str, Any]):
+  def _setup_env(self, params_config: dict[str, Any], environ_config: dict[str, Any]):
     for k, v in environ_config.items():
       if len(v) != 0:
         os.environ[k] = v
@@ -202,7 +203,7 @@ class ProcessContainer:
 
     self.environ_config = environ_config
 
-  def _setup_vision_ipc(self, all_msgs: LogIterable, frs: Dict[str, Any]):
+  def _setup_vision_ipc(self, all_msgs: LogIterable, frs: dict[str, Any]):
     assert len(self.cfg.vision_pubs) != 0
 
     vipc_server = VisionIpcServer("camerad")
@@ -223,9 +224,9 @@ class ProcessContainer:
     self.process.start()
 
   def start(
-    self, params_config: Dict[str, Any], environ_config: Dict[str, Any],
+    self, params_config: dict[str, Any], environ_config: dict[str, Any],
-    all_msgs: LogIterable, frs: Optional[Dict[str, BaseFrameReader]],
+    all_msgs: LogIterable, frs: dict[str, BaseFrameReader] | None,
-    fingerprint: Optional[str], capture_output: bool
+    fingerprint: str | None, capture_output: bool
   ):
     with self.prefix as p:
       self._setup_env(params_config, environ_config)
@@ -266,7 +267,7 @@ class ProcessContainer:
       self.prefix.clean_dirs()
       self._clean_env()
 
-  def run_step(self, msg: capnp._DynamicStructReader, frs: Optional[Dict[str, BaseFrameReader]]) -> List[capnp._DynamicStructReader]:
+  def run_step(self, msg: capnp._DynamicStructReader, frs: dict[str, BaseFrameReader] | None) -> list[capnp._DynamicStructReader]:
     assert self.rc and self.pm and self.sockets and self.process.proc
 
     output_msgs = []
@@ -580,7 +581,7 @@ def get_process_config(name: str) -> ProcessConfig:
     raise Exception(f"Cannot find process config with name: {name}") from ex
 
 
-def get_custom_params_from_lr(lr: LogIterable, initial_state: str = "first") -> Dict[str, Any]:
+def get_custom_params_from_lr(lr: LogIterable, initial_state: str = "first") -> dict[str, Any]:
   """
   Use this to get custom params dict based on provided logs.
   Useful when replaying following processes: calibrationd, paramsd, torqued
@@ -614,7 +615,7 @@ def get_custom_params_from_lr(lr: LogIterable, initial_state: str = "first") ->
   return custom_params
 
 
-def replay_process_with_name(name: Union[str, Iterable[str]], lr: LogIterable, *args, **kwargs) -> List[capnp._DynamicStructReader]:
+def replay_process_with_name(name: str | Iterable[str], lr: LogIterable, *args, **kwargs) -> list[capnp._DynamicStructReader]:
   if isinstance(name, str):
     cfgs = [get_process_config(name)]
   elif isinstance(name, Iterable):
@@ -626,10 +627,10 @@ def replay_process_with_name(name: Union[str, Iterable[str]], lr: LogIterable, *
 
 
 def replay_process(
-  cfg: Union[ProcessConfig, Iterable[ProcessConfig]], lr: LogIterable, frs: Optional[Dict[str, BaseFrameReader]] = None,
+  cfg: ProcessConfig | Iterable[ProcessConfig], lr: LogIterable, frs: dict[str, BaseFrameReader] | None = None,
-  fingerprint: Optional[str] = None, return_all_logs: bool = False, custom_params: Optional[Dict[str, Any]] = None,
+  fingerprint: str | None = None, return_all_logs: bool = False, custom_params: dict[str, Any] | None = None,
-  captured_output_store: Optional[Dict[str, Dict[str, str]]] = None, disable_progress: bool = False
+  captured_output_store: dict[str, dict[str, str]] | None = None, disable_progress: bool = False
-) -> List[capnp._DynamicStructReader]:
+) -> list[capnp._DynamicStructReader]:
   if isinstance(cfg, Iterable):
     cfgs = list(cfg)
   else:
@@ -654,9 +655,9 @@ def replay_process(
 
 
 def _replay_multi_process(
-  cfgs: List[ProcessConfig], lr: LogIterable, frs: Optional[Dict[str, BaseFrameReader]], fingerprint: Optional[str],
+  cfgs: list[ProcessConfig], lr: LogIterable, frs: dict[str, BaseFrameReader] | None, fingerprint: str | None,
-  custom_params: Optional[Dict[str, Any]], captured_output_store: Optional[Dict[str, Dict[str, str]]], disable_progress: bool
+  custom_params: dict[str, Any] | None, captured_output_store: dict[str, dict[str, str]] | None, disable_progress: bool
-) -> List[capnp._DynamicStructReader]:
+) -> list[capnp._DynamicStructReader]:
   if fingerprint is not None:
     params_config = generate_params_config(lr=lr, fingerprint=fingerprint, custom_params=custom_params)
     env_config = generate_environ_config(fingerprint=fingerprint)
@@ -690,10 +691,10 @@ def _replay_multi_process(
 
     pub_msgs = [msg for msg in all_msgs if msg.which() in lr_pubs]
     # external queue for messages taken from logs; internal queue for messages generated by processes, which will be republished
-    external_pub_queue: List[capnp._DynamicStructReader] = pub_msgs.copy()
+    external_pub_queue: list[capnp._DynamicStructReader] = pub_msgs.copy()
-    internal_pub_queue: List[capnp._DynamicStructReader] = []
+    internal_pub_queue: list[capnp._DynamicStructReader] = []
     # heap for maintaining the order of messages generated by processes, where each element: (logMonoTime, index in internal_pub_queue)
-    internal_pub_index_heap: List[Tuple[int, int]] = []
+    internal_pub_index_heap: list[tuple[int, int]] = []
 
     pbar = tqdm(total=len(external_pub_queue), disable=disable_progress)
     while len(external_pub_queue) != 0 or (len(internal_pub_index_heap) != 0 and not all(c.has_empty_queue for c in containers)):
@@ -723,7 +724,7 @@ def _replay_multi_process(
   return log_msgs
 
 
-def generate_params_config(lr=None, CP=None, fingerprint=None, custom_params=None) -> Dict[str, Any]:
+def generate_params_config(lr=None, CP=None, fingerprint=None, custom_params=None) -> dict[str, Any]:
   params_dict = {
     "OpenpilotEnabledToggle": True,
     "DisengageOnAccelerator": True,
@@ -755,7 +756,7 @@ def generate_params_config(lr=None, CP=None, fingerprint=None, custom_params=Non
   return params_dict
 
 
-def generate_environ_config(CP=None, fingerprint=None, log_dir=None) -> Dict[str, Any]:
+def generate_environ_config(CP=None, fingerprint=None, log_dir=None) -> dict[str, Any]:
   environ_dict = {}
   if platform.system() != "Darwin":
     environ_dict["PARAMS_ROOT"] = "/dev/shm/params"

selfdrive/test/process_replay/regen.py

@@ -5,7 +5,8 @@ import time
 import capnp
 import numpy as np
 
-from typing import Union, Iterable, Optional, List, Any, Dict, Tuple
+from typing import Any
+from collections.abc import Iterable
 
 from openpilot.selfdrive.test.process_replay.process_replay import CONFIGS, FAKEDATA, ProcessConfig, replay_process, get_process_config, \
                                                                    check_openpilot_enabled, get_custom_params_from_lr
@@ -40,9 +41,9 @@ class DummyFrameReader(BaseFrameReader):
 
 
 def regen_segment(
-  lr: LogIterable, frs: Optional[Dict[str, Any]] = None,
+  lr: LogIterable, frs: dict[str, Any] | None = None,
   processes: Iterable[ProcessConfig] = CONFIGS, disable_tqdm: bool = False
-) -> List[capnp._DynamicStructReader]:
+) -> list[capnp._DynamicStructReader]:
   all_msgs = sorted(lr, key=lambda m: m.logMonoTime)
   custom_params = get_custom_params_from_lr(all_msgs)
 
@@ -57,7 +58,7 @@ def regen_segment(
 def setup_data_readers(
     route: str, sidx: int, use_route_meta: bool,
     needs_driver_cam: bool = True, needs_road_cam: bool = True, dummy_driver_cam: bool = False
-) -> Tuple[LogReader, Dict[str, Any]]:
+) -> tuple[LogReader, dict[str, Any]]:
   if use_route_meta:
     r = Route(route)
     lr = LogReader(r.log_paths()[sidx])
@@ -92,7 +93,7 @@ def setup_data_readers(
 
 
 def regen_and_save(
-  route: str, sidx: int, processes: Union[str, Iterable[str]] = "all", outdir: str = FAKEDATA,
+  route: str, sidx: int, processes: str | Iterable[str] = "all", outdir: str = FAKEDATA,
   upload: bool = False, use_route_meta: bool = False, disable_tqdm: bool = False, dummy_driver_cam: bool = False
 ) -> str:
   if not isinstance(processes, str) and not hasattr(processes, "__iter__"):

selfdrive/test/process_replay/test_debayer.py

@@ -30,7 +30,7 @@ def get_frame_fn(ref_commit, test_route, tici=True):
 
 
 def bzip_frames(frames):
-  data = bytes()
+  data = b''
   for y, u, v in frames:
     data += y.tobytes()
     data += u.tobytes()

selfdrive/test/process_replay/test_processes.py

@@ -5,7 +5,7 @@ import os
 import sys
 from collections import defaultdict
 from tqdm import tqdm
-from typing import Any, DefaultDict, Dict
+from typing import Any
 
 from openpilot.selfdrive.car.car_helpers import interface_names
 from openpilot.tools.lib.openpilotci import get_url, upload_file
@@ -172,11 +172,11 @@ if __name__ == "__main__":
     untested = (set(interface_names) - set(excluded_interfaces)) - {c.lower() for c in tested_cars}
     assert len(untested) == 0, f"Cars missing routes: {str(untested)}"
 
-  log_paths: DefaultDict[str, Dict[str, Dict[str, str]]] = defaultdict(lambda: defaultdict(dict))
+  log_paths: defaultdict[str, dict[str, dict[str, str]]] = defaultdict(lambda: defaultdict(dict))
   with concurrent.futures.ProcessPoolExecutor(max_workers=args.jobs) as pool:
     if not args.upload_only:
       download_segments = [seg for car, seg in segments if car in tested_cars]
-      log_data: Dict[str, LogReader] = {}
+      log_data: dict[str, LogReader] = {}
       p1 = pool.map(get_log_data, download_segments)
       for segment, lr in tqdm(p1, desc="Getting Logs", total=len(download_segments)):
         log_data[segment] = lr

selfdrive/test/update_ci_routes.py

@@ -3,7 +3,7 @@ import os
 import re
 import subprocess
 import sys
-from typing import Iterable, List, Optional
+from collections.abc import Iterable
 
 from tqdm import tqdm
 
@@ -12,14 +12,14 @@ from openpilot.selfdrive.test.process_replay.test_processes import source_segmen
 from openpilot.tools.lib.azure_container import AzureContainer
 from openpilot.tools.lib.openpilotcontainers import DataCIContainer, DataProdContainer, OpenpilotCIContainer
 
-SOURCES: List[AzureContainer] = [
+SOURCES: list[AzureContainer] = [
   DataProdContainer,
   DataCIContainer
 ]
 
 DEST = OpenpilotCIContainer
 
-def upload_route(path: str, exclude_patterns: Optional[Iterable[str]] = None) -> None:
+def upload_route(path: str, exclude_patterns: Iterable[str] | None = None) -> None:
   if exclude_patterns is None:
     exclude_patterns = [r'dcamera\.hevc']
 

selfdrive/thermald/power_monitoring.py

@@ -1,6 +1,5 @@
 import time
 import threading
-from typing import Optional
 
 from openpilot.common.params import Params
 from openpilot.system.hardware import HARDWARE
@@ -38,7 +37,7 @@ class PowerMonitoring:
     self.car_battery_capacity_uWh = max((CAR_BATTERY_CAPACITY_uWh / 10), int(car_battery_capacity_uWh))
 
   # Calculation tick
-  def calculate(self, voltage: Optional[int], ignition: bool):
+  def calculate(self, voltage: int | None, ignition: bool):
     try:
       now = time.monotonic()
 
@@ -108,7 +107,7 @@ class PowerMonitoring:
     return int(self.car_battery_capacity_uWh)
 
   # See if we need to shutdown
-  def should_shutdown(self, ignition: bool, in_car: bool, offroad_timestamp: Optional[float], started_seen: bool):
+  def should_shutdown(self, ignition: bool, in_car: bool, offroad_timestamp: float | None, started_seen: bool):
     if offroad_timestamp is None:
       return False
 

selfdrive/thermald/thermald.py

@@ -6,7 +6,6 @@ import threading
 import time
 from collections import OrderedDict, namedtuple
 from pathlib import Path
-from typing import Dict, Optional, Tuple
 
 import psutil
 
@@ -50,9 +49,9 @@ THERMAL_BANDS = OrderedDict({
 # Override to highest thermal band when offroad and above this temp
 OFFROAD_DANGER_TEMP = 75
 
-prev_offroad_states: Dict[str, Tuple[bool, Optional[str]]] = {}
+prev_offroad_states: dict[str, tuple[bool, str | None]] = {}
 
-tz_by_type: Optional[Dict[str, int]] = None
+tz_by_type: dict[str, int] | None = None
 def populate_tz_by_type():
   global tz_by_type
   tz_by_type = {}
@@ -87,7 +86,7 @@ def read_thermal(thermal_config):
   return dat
 
 
-def set_offroad_alert_if_changed(offroad_alert: str, show_alert: bool, extra_text: Optional[str]=None):
+def set_offroad_alert_if_changed(offroad_alert: str, show_alert: bool, extra_text: str | None=None):
   if prev_offroad_states.get(offroad_alert, None) == (show_alert, extra_text):
     return
   prev_offroad_states[offroad_alert] = (show_alert, extra_text)
@@ -169,16 +168,16 @@ def thermald_thread(end_event, hw_queue) -> None:
 
   count = 0
 
-  onroad_conditions: Dict[str, bool] = {
+  onroad_conditions: dict[str, bool] = {
     "ignition": False,
   }
-  startup_conditions: Dict[str, bool] = {}
+  startup_conditions: dict[str, bool] = {}
-  startup_conditions_prev: Dict[str, bool] = {}
+  startup_conditions_prev: dict[str, bool] = {}
 
-  off_ts: Optional[float] = None
+  off_ts: float | None = None
-  started_ts: Optional[float] = None
+  started_ts: float | None = None
   started_seen = False
-  startup_blocked_ts: Optional[float] = None
+  startup_blocked_ts: float | None = None
   thermal_status = ThermalStatus.yellow
 
   last_hw_state = HardwareState(

selfdrive/ui/soundd.py

@@ -3,7 +3,6 @@ import numpy as np
 import time
 import wave
 
-from typing import Dict, Optional, Tuple
 
 from cereal import car, messaging
 from openpilot.common.basedir import BASEDIR
@@ -27,7 +26,7 @@ DB_SCALE = 30 # AMBIENT_DB + DB_SCALE is where MAX_VOLUME is applied
 AudibleAlert = car.CarControl.HUDControl.AudibleAlert
 
 
-sound_list: Dict[int, Tuple[str, Optional[int], float]] = {
+sound_list: dict[int, tuple[str, int | None, float]] = {
   # AudibleAlert, file name, play count (none for infinite)
   AudibleAlert.engage: ("engage.wav", 1, MAX_VOLUME),
   AudibleAlert.disengage: ("disengage.wav", 1, MAX_VOLUME),
@@ -64,7 +63,7 @@ class Soundd:
     self.spl_filter_weighted = FirstOrderFilter(0, 2.5, FILTER_DT, initialized=False)
 
   def load_sounds(self):
-    self.loaded_sounds: Dict[int, np.ndarray] = {}
+    self.loaded_sounds: dict[int, np.ndarray] = {}
 
     # Load all sounds
     for sound in sound_list:

selfdrive/ui/tests/test_translations.py

@@ -12,7 +12,7 @@ from parameterized import parameterized_class
 
 from openpilot.selfdrive.ui.update_translations import TRANSLATIONS_DIR, LANGUAGES_FILE, update_translations
 
-with open(LANGUAGES_FILE, "r") as f:
+with open(LANGUAGES_FILE) as f:
   translation_files = json.load(f)
 
 UNFINISHED_TRANSLATION_TAG = "<translation type=\"unfinished\""  # non-empty translations can be marked unfinished
@@ -28,7 +28,7 @@ class TestTranslations(unittest.TestCase):
   @staticmethod
   def _read_translation_file(path, file):
     tr_file = os.path.join(path, f"{file}.ts")
-    with open(tr_file, "r") as f:
+    with open(tr_file) as f:
       return f.read()
 
   def test_missing_translation_files(self):
@@ -83,7 +83,7 @@ class TestTranslations(unittest.TestCase):
           for nf in numerusform:
             self.assertIsNotNone(nf, f"Ensure all plural translation forms are completed: {source_text}")
             self.assertIn("%n", nf, "Ensure numerus argument (%n) exists in translation.")
-            self.assertIsNone(FORMAT_ARG.search(nf), "Plural translations must use %n, not %1, %2, etc.: {}".format(numerusform))
+            self.assertIsNone(FORMAT_ARG.search(nf), f"Plural translations must use %n, not %1, %2, etc.: {numerusform}")
 
         else:
           self.assertIsNotNone(translation.text, f"Ensure translation is completed: {source_text}")

selfdrive/ui/translations/create_badges.py

@@ -13,13 +13,13 @@ BADGE_HEIGHT = 20 + 8
 SHIELDS_URL = "https://img.shields.io/badge"
 
 if __name__ == "__main__":
-  with open(LANGUAGES_FILE, "r") as f:
+  with open(LANGUAGES_FILE) as f:
     translation_files = json.load(f)
 
   badge_svg = []
   max_badge_width = 0  # keep track of max width to set parent element
   for idx, (name, file) in enumerate(translation_files.items()):
-    with open(os.path.join(TRANSLATIONS_DIR, f"{file}.ts"), "r") as tr_f:
+    with open(os.path.join(TRANSLATIONS_DIR, f"{file}.ts")) as tr_f:
       tr_file = tr_f.read()
 
     total_translations = 0

selfdrive/ui/update_translations.py

@@ -28,7 +28,7 @@ def update_translations(vanish: bool = False, translation_files: None | list[str
   generate_translations_include()
 
   if translation_files is None:
-    with open(LANGUAGES_FILE, "r") as f:
+    with open(LANGUAGES_FILE) as f:
       translation_files = json.load(f).values()
 
   for file in translation_files:

selfdrive/updated.py

@@ -11,7 +11,6 @@ import time
 import threading
 from collections import defaultdict
 from pathlib import Path
-from typing import List, Union, Optional
 from markdown_it import MarkdownIt
 
 from openpilot.common.basedir import BASEDIR
@@ -64,7 +63,7 @@ def write_time_to_param(params, param) -> None:
   t = datetime.datetime.utcnow()
   params.put(param, t.isoformat().encode('utf8'))
 
-def read_time_from_param(params, param) -> Optional[datetime.datetime]:
+def read_time_from_param(params, param) -> datetime.datetime | None:
   t = params.get(param, encoding='utf8')
   try:
     return datetime.datetime.fromisoformat(t)
@@ -72,7 +71,7 @@ def read_time_from_param(params, param) -> Optional[datetime.datetime]:
     pass
   return None
 
-def run(cmd: List[str], cwd: Optional[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')
 
 
@@ -234,7 +233,7 @@ def handle_agnos_update() -> None:
 class Updater:
   def __init__(self):
     self.params = Params()
-    self.branches = defaultdict(lambda: '')
+    self.branches = defaultdict(str)
     self._has_internet: bool = False
 
   @property
@@ -243,7 +242,7 @@ class Updater:
 
   @property
   def target_branch(self) -> str:
-    b: Union[str, None] = self.params.get("UpdaterTargetBranch", encoding='utf-8')
+    b: str | None = self.params.get("UpdaterTargetBranch", encoding='utf-8')
     if b is None:
       b = self.get_branch(BASEDIR)
     return b
@@ -272,7 +271,7 @@ class Updater:
   def get_commit_hash(self, path: str = OVERLAY_MERGED) -> str:
     return run(["git", "rev-parse", "HEAD"], path).rstrip()
 
-  def set_params(self, update_success: bool, failed_count: int, exception: Optional[str]) -> None:
+  def set_params(self, update_success: bool, failed_count: int, exception: str | None) -> None:
     self.params.put("UpdateFailedCount", str(failed_count))
     self.params.put("UpdaterTargetBranch", self.target_branch)
 

system/hardware/base.py

@@ -1,6 +1,5 @@
 from abc import abstractmethod, ABC
 from collections import namedtuple
-from typing import Dict
 
 from cereal import log
 
@@ -10,7 +9,7 @@ NetworkType = log.DeviceState.NetworkType
 
 class HardwareBase(ABC):
   @staticmethod
-  def get_cmdline() -> Dict[str, str]:
+  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}

system/hardware/tici/agnos.py

@@ -6,7 +6,7 @@ import os
 import struct
 import subprocess
 import time
-from typing import Dict, Generator, List, Tuple, Union
+from collections.abc import Generator
 
 import requests
 
@@ -117,7 +117,7 @@ def get_raw_hash(path: str, partition_size: int) -> str:
   return raw_hash.hexdigest().lower()
 
 
-def verify_partition(target_slot_number: int, partition: Dict[str, Union[str, int]], force_full_check: bool = False) -> bool:
+def verify_partition(target_slot_number: int, partition: dict[str, str | int], force_full_check: bool = False) -> bool:
   full_check = partition['full_check'] or force_full_check
   path = get_partition_path(target_slot_number, partition)
 
@@ -184,7 +184,7 @@ def extract_casync_image(target_slot_number: int, partition: dict, cloudlog):
 
   target = casync.parse_caibx(partition['casync_caibx'])
 
-  sources: List[Tuple[str, casync.ChunkReader, casync.ChunkDict]] = []
+  sources: list[tuple[str, casync.ChunkReader, casync.ChunkDict]] = []
 
   # First source is the current partition.
   try:

system/hardware/tici/amplifier.py

@@ -2,7 +2,6 @@
 import time
 from smbus2 import SMBus
 from collections import namedtuple
-from typing import List
 
 # https://datasheets.maximintegrated.com/en/ds/MAX98089.pdf
 
@@ -110,7 +109,7 @@ class Amplifier:
   def _get_shutdown_config(self, amp_disabled: bool) -> AmpConfig:
     return AmpConfig("Global shutdown", 0b0 if amp_disabled else 0b1, 0x51, 7, 0b10000000)
 
-  def _set_configs(self, configs: List[AmpConfig]) -> None:
+  def _set_configs(self, configs: list[AmpConfig]) -> None:
     with SMBus(self.AMP_I2C_BUS) as bus:
       for config in configs:
         if self.debug:
@@ -123,7 +122,7 @@ class Amplifier:
         if self.debug:
           print(f"  Changed {hex(config.register)}: {hex(old_value)} -> {hex(new_value)}")
 
-  def set_configs(self, configs: List[AmpConfig]) -> bool:
+  def set_configs(self, configs: list[AmpConfig]) -> bool:
     # retry in case panda is using the amp
     tries = 15
     for i in range(15):

system/hardware/tici/casync.py

@@ -7,7 +7,7 @@ import sys
 import time
 from abc import ABC, abstractmethod
 from collections import defaultdict, namedtuple
-from typing import Callable, Dict, List, Optional, Tuple
+from collections.abc import Callable
 
 import requests
 from Crypto.Hash import SHA512
@@ -28,7 +28,7 @@ CHUNK_DOWNLOAD_RETRIES = 3
 CAIBX_DOWNLOAD_TIMEOUT = 120
 
 Chunk = namedtuple('Chunk', ['sha', 'offset', 'length'])
-ChunkDict = Dict[bytes, Chunk]
+ChunkDict = dict[bytes, Chunk]
 
 
 class ChunkReader(ABC):
@@ -83,7 +83,7 @@ class RemoteChunkReader(ChunkReader):
     return decompressor.decompress(contents)
 
 
-def parse_caibx(caibx_path: str) -> List[Chunk]:
+def parse_caibx(caibx_path: str) -> list[Chunk]:
   """Parses the chunks from a caibx file. Can handle both local and remote files.
   Returns a list of chunks with hash, offset and length"""
   caibx: io.BufferedIOBase
@@ -132,7 +132,7 @@ def parse_caibx(caibx_path: str) -> List[Chunk]:
   return chunks
 
 
-def build_chunk_dict(chunks: List[Chunk]) -> ChunkDict:
+def build_chunk_dict(chunks: list[Chunk]) -> ChunkDict:
   """Turn a list of chunks into a dict for faster lookups based on hash.
   Keep first chunk since it's more likely to be already downloaded."""
   r = {}
@@ -142,11 +142,11 @@ def build_chunk_dict(chunks: List[Chunk]) -> ChunkDict:
   return r
 
 
-def extract(target: List[Chunk],
+def extract(target: list[Chunk],
-            sources: List[Tuple[str, ChunkReader, ChunkDict]],
+            sources: list[tuple[str, ChunkReader, ChunkDict]],
             out_path: str,
-            progress: Optional[Callable[[int], None]] = None):
+            progress: Callable[[int], None] | None = None):
-  stats: Dict[str, int] = defaultdict(int)
+  stats: dict[str, int] = defaultdict(int)
 
   mode = 'rb+' if os.path.exists(out_path) else 'wb'
   with open(out_path, mode) as out:
@@ -181,7 +181,7 @@ def extract(target: List[Chunk],
   return stats
 
 
-def print_stats(stats: Dict[str, int]):
+def print_stats(stats: dict[str, int]):
   total_bytes = sum(stats.values())
   print(f"Total size: {total_bytes / 1024 / 1024:.2f} MB")
   for name, total in stats.items():

system/hardware/tici/power_monitor.py

@@ -3,7 +3,6 @@ import sys
 import time
 import datetime
 import numpy as np
-from typing import List
 from collections import deque
 
 from openpilot.common.realtime import Ratekeeper
@@ -14,7 +13,7 @@ def read_power():
   with open("/sys/bus/i2c/devices/0-0040/hwmon/hwmon1/power1_input") as f:
     return int(f.read()) / 1e6
 
-def sample_power(seconds=5) -> List[float]:
+def sample_power(seconds=5) -> list[float]:
   rate = 123
   rk = Ratekeeper(rate, print_delay_threshold=None)
 

system/hardware/tici/precise_power_measure.py

@@ -6,4 +6,4 @@ if __name__ == '__main__':
   print("measuring for 5 seconds")
   for _ in range(3):
     pwrs = sample_power()
-    print("mean %.2f std %.2f" % (np.mean(pwrs), np.std(pwrs)))
+    print(f"mean {np.mean(pwrs):.2f} std {np.std(pwrs):.2f}")

system/hardware/tici/tests/compare_casync_manifest.py

@@ -3,7 +3,6 @@ import argparse
 import collections
 import multiprocessing
 import os
-from typing import Dict, List
 
 import requests
 from tqdm import tqdm
@@ -42,7 +41,7 @@ if __name__ == "__main__":
     szs = list(tqdm(pool.imap(get_chunk_download_size, to), total=len(to)))
   chunk_sizes = {t.sha: sz for (t, sz) in zip(to, szs, strict=True)}
 
-  sources: Dict[str, List[int]] = {
+  sources: dict[str, list[int]] = {
     'seed': [],
     'remote_uncompressed': [],
     'remote_compressed': [],

system/hardware/tici/tests/test_power_draw.py

@@ -7,7 +7,6 @@ import time
 import numpy as np
 from dataclasses import dataclass
 from tabulate import tabulate
-from typing import List
 
 import cereal.messaging as messaging
 from cereal.services import SERVICE_LIST
@@ -22,9 +21,9 @@ MAX_WARMUP_TIME = 30  # seconds to wait for SAMPLE_TIME consecutive valid sample
 
 @dataclass
 class Proc:
-  procs: List[str]
+  procs: list[str]
   power: float
-  msgs: List[str]
+  msgs: list[str]
   rtol: float = 0.05
   atol: float = 0.12
 
@@ -66,7 +65,7 @@ class TestPowerDraw(unittest.TestCase):
     return np.core.numeric.isclose(used, proc.power, rtol=proc.rtol, atol=proc.atol)
 
   def tabulate_msg_counts(self, msgs_and_power):
-    msg_counts = defaultdict(lambda: 0)
+    msg_counts = defaultdict(int)
     for _, counts in msgs_and_power:
       for msg, count in counts.items():
         msg_counts[msg] += count

system/loggerd/deleter.py

@@ -2,7 +2,6 @@
 import os
 import shutil
 import threading
-from typing import List
 from openpilot.system.hardware.hw import Paths
 from openpilot.common.swaglog import cloudlog
 from openpilot.system.loggerd.config import get_available_bytes, get_available_percent
@@ -23,7 +22,7 @@ def has_preserve_xattr(d: str) -> bool:
   return getxattr(os.path.join(Paths.log_root(), d), PRESERVE_ATTR_NAME) == PRESERVE_ATTR_VALUE
 
 
-def get_preserved_segments(dirs_by_creation: List[str]) -> List[str]:
+def get_preserved_segments(dirs_by_creation: list[str]) -> list[str]:
   preserved = []
   for n, d in enumerate(filter(has_preserve_xattr, reversed(dirs_by_creation))):
     if n == PRESERVE_COUNT:

system/loggerd/tests/loggerd_tests_common.py

@@ -2,7 +2,6 @@ import os
 import random
 import unittest
 from pathlib import Path
-from typing import Optional
 
 
 import openpilot.system.loggerd.deleter as deleter
@@ -12,7 +11,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: Optional[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:
@@ -82,7 +81,7 @@ class UploaderTestCase(unittest.TestCase):
     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: Optional[bytes] = None, preserve_xattr: Optional[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_deleter.py

@@ -4,7 +4,7 @@ import threading
 import unittest
 from collections import namedtuple
 from pathlib import Path
-from typing import Sequence
+from collections.abc import Sequence
 
 import openpilot.system.loggerd.deleter as deleter
 from openpilot.common.timeout import Timeout, TimeoutException

system/loggerd/tests/test_loggerd.py

@@ -8,7 +8,6 @@ import subprocess
 import time
 from collections import defaultdict
 from pathlib import Path
-from typing import Dict, List
 from flaky import flaky
 
 import cereal.messaging as messaging
@@ -76,7 +75,7 @@ class TestLoggerd:
     end_type = SentinelType.endOfRoute if route else SentinelType.endOfSegment
     assert msgs[-1].sentinel.type == end_type
 
-  def _publish_random_messages(self, services: List[str]) -> Dict[str, list]:
+  def _publish_random_messages(self, services: list[str]) -> dict[str, list]:
     pm = messaging.PubMaster(services)
 
     managed_processes["loggerd"].start()

system/loggerd/tests/test_uploader.py

@@ -6,7 +6,6 @@ import unittest
 import logging
 import json
 from pathlib import Path
-from typing import List, Optional
 from openpilot.system.hardware.hw import Paths
 
 from openpilot.common.swaglog import cloudlog
@@ -53,7 +52,7 @@ class TestUploader(UploaderTestCase):
     self.end_event.set()
     self.up_thread.join()
 
-  def gen_files(self, lock=False, xattr: Optional[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))
@@ -62,7 +61,7 @@ class TestUploader(UploaderTestCase):
       f_paths.append(self.make_file_with_data("boot", f"{self.seg_dir}", 1, lock=lock, upload_xattr=xattr))
     return f_paths
 
-  def gen_order(self, seg1: List[int], seg2: List[int], boot=True) -> List[str]:
+  def gen_order(self, seg1: list[int], seg2: list[int], boot=True) -> list[str]:
     keys = []
     if boot:
       keys += [f"boot/{self.seg_format.format(i)}.bz2" for i in seg1]

system/loggerd/uploader.py

@@ -9,7 +9,8 @@ import threading
 import time
 import traceback
 import datetime
-from typing import BinaryIO, Iterator, List, Optional, Tuple
+from typing import BinaryIO
+from collections.abc import Iterator
 
 from cereal import log
 import cereal.messaging as messaging
@@ -42,10 +43,10 @@ class FakeResponse:
     self.request = FakeRequest()
 
 
-def get_directory_sort(d: str) -> List[str]:
+def get_directory_sort(d: str) -> list[str]:
   return [s.rjust(10, '0') for s in d.rsplit('--', 1)]
 
-def listdir_by_creation(d: str) -> List[str]:
+def listdir_by_creation(d: str) -> list[str]:
   if not os.path.isdir(d):
     return []
 
@@ -82,7 +83,7 @@ class Uploader:
     self.immediate_folders = ["crash/", "boot/"]
     self.immediate_priority = {"qlog": 0, "qlog.bz2": 0, "qcamera.ts": 1}
 
-  def list_upload_files(self, metered: bool) -> Iterator[Tuple[str, str, str]]:
+  def list_upload_files(self, metered: bool) -> Iterator[tuple[str, str, str]]:
     r = self.params.get("AthenadRecentlyViewedRoutes", encoding="utf8")
     requested_routes = [] if r is None else r.split(",")
 
@@ -121,7 +122,7 @@ class Uploader:
 
         yield name, key, fn
 
-  def next_file_to_upload(self, metered: bool) -> Optional[Tuple[str, str, str]]:
+  def next_file_to_upload(self, metered: bool) -> tuple[str, str, str] | None:
     upload_files = list(self.list_upload_files(metered))
 
     for name, key, fn in upload_files:
@@ -207,7 +208,7 @@ class Uploader:
     return success
 
 
-  def step(self, network_type: int, metered: bool) -> Optional[bool]:
+  def step(self, network_type: int, metered: bool) -> bool | None:
     d = self.next_file_to_upload(metered)
     if d is None:
       return None
@@ -221,7 +222,7 @@ class Uploader:
     return self.upload(name, key, fn, network_type, metered)
 
 
-def main(exit_event: Optional[threading.Event] = None) -> None:
+def main(exit_event: threading.Event | None = None) -> None:
   if exit_event is None:
     exit_event = threading.Event()
 

system/loggerd/xattr_cache.py

@@ -1,10 +1,9 @@
 import os
 import errno
-from typing import Dict, Optional, Tuple
 
-_cached_attributes: Dict[Tuple, Optional[bytes]] = {}
+_cached_attributes: dict[tuple, bytes | None] = {}
 
-def getxattr(path: str, attr_name: str) -> Optional[bytes]:
+def getxattr(path: str, attr_name: str) -> bytes | None:
   key = (path, attr_name)
   if key not in _cached_attributes:
     try:

system/qcomgpsd/nmeaport.py

@@ -93,7 +93,7 @@ def nmea_checksum_ok(s):
 def process_nmea_port_messages(device:str="/dev/ttyUSB1") -> NoReturn:
   while True:
     try:
-      with open(device, "r") as nmeaport:
+      with open(device) as nmeaport:
         for line in nmeaport:
           line = line.strip()
           if DEBUG:

system/qcomgpsd/qcomgpsd.py

@@ -10,7 +10,7 @@ import shutil
 import subprocess
 import datetime
 from multiprocessing import Process, Event
-from typing import NoReturn, Optional
+from typing import NoReturn
 from struct import unpack_from, calcsize, pack
 
 from cereal import log
@@ -90,7 +90,7 @@ def try_setup_logs(diag, logs):
   return setup_logs(diag, logs)
 
 @retry(attempts=3, delay=1.0)
-def at_cmd(cmd: str) -> Optional[str]:
+def at_cmd(cmd: str) -> str | None:
   return subprocess.check_output(f"mmcli -m any --timeout 30 --command='{cmd}'", shell=True, encoding='utf8')
 
 def gps_enabled() -> bool:
@@ -342,7 +342,7 @@ def main() -> NoReturn:
       gps.bearingDeg = report["q_FltHeadingRad"] * 180/math.pi
 
       # TODO needs update if there is another leap second, after june 2024?
-      dt_timestamp = (datetime.datetime(1980, 1, 6, 0, 0, 0, 0, datetime.timezone.utc) +
+      dt_timestamp = (datetime.datetime(1980, 1, 6, 0, 0, 0, 0, datetime.UTC) +
                       datetime.timedelta(weeks=report['w_GpsWeekNumber']) +
                       datetime.timedelta(seconds=(1e-3*report['q_GpsFixTimeMs'] - 18)))
       gps.unixTimestampMillis = dt_timestamp.timestamp()*1e3