mawei
/
openpilot0
mirror of https://github.com/commaai/openpilot.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Cameron Clough 2024-02-25 00:41:23 +00:00 committed by GitHub
parent 6ed72b3626
commit 995250ae49
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
125 changed files with 464 additions and 525 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
common/file_helpers.py
View File

@ -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)

7
common/gpio.py
View File

@ -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():

5
common/mock/__init__.py
View File

@ -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]

3
common/prefix.py
View File

@ -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

7
common/realtime.py
View File

@ -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

2
common/transformations/orientation.py
View File

@ -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,

3
openpilot/__init__.py
View File

@ -1,3 +0,0 @@

57
selfdrive/athena/athenad.py
View File

@ -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]]
UploadItemDict = Dict[str, Union[str, bool, int, float, Dict[str, str]]]
UploadFileDict = dict[str, str | int | float | bool]
UploadItemDict = dict[str, str | bool | int | float | dict[str, str]]
UploadFilesToUrlResponse = Dict[str, Union[int, List[UploadItemDict], List[str]]]
UploadFilesToUrlResponse = dict[str, int | list[UploadItemDict] | list[str]]
@dataclass
class UploadFile:
fn: str
url: str
headers: Dict[str, str]
headers: dict[str, str]
allow_cellular: bool
@classmethod
@ -77,9 +78,9 @@ 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] = []
failed: List[str] = []
items: list[UploadItemDict] = []
failed: list[str] = []
for file in files:
if len(file.fn) == 0 or file.fn[0] == '/' or '..' in file.fn or len(file.url) == 0:
failed.append(file.fn)
@ -418,13 +419,13 @@ def uploadFilesToUrls(files_data: List[UploadFileDict]) -> UploadFilesToUrlRespo
@dispatcher.add_method
def listUploadQueue() -> List[UploadItemDict]:
def listUploadQueue() -> list[UploadItemDict]:
items = list(upload_queue.queue) + list(cur_upload_items.values())
return [asdict(i) for i in items if (i is not None) and (i.id not in cancelled_uploads)]
@dispatcher.add_method
def cancelUpload(upload_id: Union[str, List[str]]) -> Dict[str, Union[int, str]]:
def cancelUpload(upload_id: str | list[str]) -> dict[str, int | str]:
if not isinstance(upload_id, list):
upload_id = [upload_id]
@ -437,7 +438,7 @@ def cancelUpload(upload_id: Union[str, List[str]]) -> Dict[str, Union[int, str]]
return {"success": 1}
@dispatcher.add_method
def setRouteViewed(route: str) -> Dict[str, Union[int, str]]:
def setRouteViewed(route: str) -> dict[str, int | str]:
# maintain a list of the last 10 routes viewed in connect
params = Params()
@ -452,7 +453,7 @@ def setRouteViewed(route: str) -> Dict[str, Union[int, str]]:
return {"success": 1}
def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local_port: int) -> Dict[str, int]:
def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local_port: int) -> dict[str, int]:
try:
if local_port not in LOCAL_PORT_WHITELIST:
raise Exception("Requested local port not whitelisted")
@ -486,7 +487,7 @@ def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local
@dispatcher.add_method
def getPublicKey() -> Optional[str]:
def getPublicKey() -> str | None:
if not os.path.isfile(Paths.persist_root() + '/comma/id_rsa.pub'):
return None
@ -526,7 +527,7 @@ def getNetworks():
@dispatcher.add_method
def takeSnapshot() -> Optional[Union[str, Dict[str, str]]]:
def takeSnapshot() -> str | dict[str, str] | None:
from openpilot.system.camerad.snapshot.snapshot import jpeg_write, snapshot
ret = snapshot()
if ret is not None:
@ -543,7 +544,7 @@ def 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:

9
selfdrive/athena/registration.py
View File

@ -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
imei2: Optional[str] = None
imei1: str | None = None
imei2: str | None = None
while imei1 is None and imei2 is None:
try:
imei1, imei2 = HARDWARE.get_imei(0), HARDWARE.get_imei(1)

3
selfdrive/athena/tests/test_athenad.py
View File

@ -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:

6
selfdrive/athena/tests/test_athenad_ping.py
View File

@ -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]:
return cast(Optional[str], self.params.get("LastAthenaPingTime", encoding="utf-8"))
def _get_ping_time(self) -> str | None:
return cast(str | None, self.params.get("LastAthenaPingTime", encoding="utf-8"))
def _clear_ping_time(self) -> None:
self.params.remove("LastAthenaPingTime")

4
selfdrive/boardd/pandad.py
View File

@ -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))

19
selfdrive/car/__init__.py
View File

@ -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],
unpressed_btn: int = 0) -> List[capnp.lib.capnp._DynamicStructBuilder]:
events: List[capnp.lib.capnp._DynamicStructBuilder] = []
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]:
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}

6
selfdrive/car/car_helpers.py
View File

@ -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

3
selfdrive/car/chrysler/values.py
View File

@ -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"),

15
selfdrive/car/docs.py
View File

@ -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]:
all_car_info: List[CarInfo] = []
def get_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:
with open(template_fn, "r") as f:
def generate_cars_md(all_car_info: list[CarInfo], template_fn: str) -> str:
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()]

33
selfdrive/car/docs_definitions.py
View File

@ -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
footnotes: List[Enum] = field(default_factory=list)
min_steer_speed: Optional[float] = None
min_enable_speed: Optional[float] = None
auto_resume: Optional[bool] = None
video_link: str | None = None
footnotes: list[Enum] = field(default_factory=list)
min_steer_speed: float | None = None
min_enable_speed: float | None = 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):

13
selfdrive/car/ecu_addrs.py
View File

@ -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],
responses: Set[EcuAddrBusType], timeout: float = 1, debug: bool = False) -> Set[EcuAddrBusType]:
ecu_responses: Set[EcuAddrBusType] = set() # set((addr, subaddr, bus),)
def get_ecu_addrs(logcan: messaging.SubSocket, sendcan: messaging.PubSocket, queries: set[EcuAddrBusType],
responses: set[EcuAddrBusType], timeout: float = 1, debug: bool = False) -> set[EcuAddrBusType]:
ecu_responses: set[EcuAddrBusType] = set() # set((addr, subaddr, bus),)
try:
msgs = [make_tester_present_msg(addr, bus, subaddr) for addr, subaddr, bus in queries]

4
selfdrive/car/ford/tests/test_ford.py
View File

@ -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")

5
selfdrive/car/ford/values.py
View File

@ -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"),

26
selfdrive/car/fw_query_definitions.py
View File

@ -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]]
EcuAddrBusType = Tuple[int, Optional[int], int]
EcuAddrSubAddr = Tuple[int, int, Optional[int]]
AddrType = tuple[int, int | None]
EcuAddrBusType = tuple[int, int | None, int]
EcuAddrSubAddr = tuple[int, int, int | None]
LiveFwVersions = Dict[AddrType, Set[bytes]]
OfflineFwVersions = Dict[str, Dict[EcuAddrSubAddr, List[bytes]]]
LiveFwVersions = dict[AddrType, set[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]
response: List[bytes]
whitelist_ecus: List[int] = field(default_factory=list)
request: list[bytes]
response: list[bytes]
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)):

27
selfdrive/car/fw_versions.py
View File

@ -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],
filter_brand: Optional[str] = None) -> Dict[AddrType, Set[bytes]]:
fw_versions_dict: DefaultDict[AddrType, Set[bytes]] = defaultdict(set)
def build_fw_dict(fw_versions: list[capnp.lib.capnp._DynamicStructBuilder],
filter_brand: str | None = None) -> dict[AddrType, set[bytes]]:
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: []}
parallel_queries: Dict[bool, List[EcuAddrBusType]] = {True: [], False: []}
queries: dict[bool, list[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()

5
selfdrive/car/gm/values.py
View File

@ -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}

3
selfdrive/car/honda/values.py
View File

@ -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),

9
selfdrive/car/hyundai/values.py
View File

@ -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)

23
selfdrive/car/interfaces.py
View File

@ -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]],
car_fw: List[car.CarParams.CarFw], experimental_long: bool, docs: bool):
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):
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'])

3
selfdrive/car/mazda/values.py
View File

@ -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"),

3
selfdrive/car/mock/values.py
View File

@ -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,
}

3
selfdrive/car/nissan/values.py
View File

@ -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

3
selfdrive/car/subaru/values.py
View File

@ -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])

3
selfdrive/car/tesla/values.py
View File

@ -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"),
}

6
selfdrive/car/tests/routes.py
View File

@ -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]
segment: Optional[int] = None
car_model: str | None
segment: int | None = None
routes = [

4
selfdrive/car/tests/test_docs.py
View File

@ -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

3
selfdrive/car/tests/test_fingerprints.py
View File

@ -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:

3
selfdrive/car/tests/test_lateral_limits.py
View File

@ -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)])

15
selfdrive/car/tests/test_models.py
View File

@ -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
test_route: Optional[CarTestRoute] = None
car_model: str | None = 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]
car_safety_mode_frame: Optional[int]
elm_frame: int | None
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):

7
selfdrive/car/toyota/values.py
View File

@ -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():

5
selfdrive/car/volkswagen/values.py
View File

@ -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"),

11
selfdrive/controls/lib/alertmanager.py
View File

@ -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:

18
selfdrive/controls/lib/events.py
View File

@ -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.static_events: List[int] = []
self.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)

3
selfdrive/controls/lib/vehicle_model.py
View File

@ -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:

12
selfdrive/controls/radard.py
View File

@ -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,
model_v_ego: float, low_speed_override: bool = True) -> Dict[str, Any]:
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]:
# 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

5
selfdrive/debug/can_print_changes.py
View File

@ -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
new_lr: Optional[LogIterable] = None
init_lr: LogIterable | None = None
new_lr: LogIterable | None = None
if args.init:
if args.init == '':

6
selfdrive/debug/check_freq.py
View File

@ -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))
valids: DefaultDict[str, Deque[bool]] = 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))
t = time.monotonic()
for name in socket_names:

3
selfdrive/debug/check_lag.py
View File

@ -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()

4
selfdrive/debug/check_timings.py
View File

@ -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:

6
selfdrive/debug/count_events.py
View File

@ -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]]] = []
alerts: List[Tuple[float, str]] = []
events: list[tuple[float, set[str]]] = []
alerts: list[tuple[float, str]] = []
start_time = math.inf
end_time = -math.inf
ignition_off = None

4
selfdrive/debug/cpu_usage_stat.py
View File

@ -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)

2
selfdrive/debug/format_fingerprints.py
View File

@ -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:

3
selfdrive/debug/internal/measure_modeld_packet_drop.py
View File

@ -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

7
selfdrive/debug/live_cpu_and_temp.py
View File

@ -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_t: Optional[int] = None
prev_proclog: capnp._DynamicStructReader | None = 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:

18
selfdrive/locationd/calibrationd.py
View File

@ -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],
rot: List[float],
wide_from_device_euler: List[float],
trans_std: List[float],
road_transform_trans: List[float],
road_transform_trans_std: List[float]) -> Optional[np.ndarray]:
def handle_cam_odom(self, trans: list[float],
rot: list[float],
wide_from_device_euler: list[float],
trans_std: list[float],
road_transform_trans: list[float],
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))

10
selfdrive/locationd/helpers.py
View File

@ -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)

4
selfdrive/locationd/models/car_kf.py
View File

@ -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),

3
selfdrive/manager/build.py
View File

@ -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)

7
selfdrive/manager/manager.py
View File

@ -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)

12
selfdrive/manager/process.py
View File

@ -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 = []

5
selfdrive/modeld/dmonitoringmodeld.py
View File

@ -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

5
selfdrive/modeld/fill_model_msg.py
View File

@ -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

3
selfdrive/modeld/get_model_metadata.py
View File

@ -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

9
selfdrive/modeld/modeld.py
View File

@ -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,

5
selfdrive/modeld/navmodeld.py
View File

@ -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()

3
selfdrive/modeld/parse_model_outputs.py
View File

@ -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))

4
selfdrive/modeld/runners/onnxmodel.py
View File

@ -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:

16
selfdrive/navd/helpers.py
View File

@ -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

6
selfdrive/statsd.py
View File

@ -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

4
selfdrive/test/ciui.py
View File

@ -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())

17
selfdrive/test/fuzzy_generation.py
View File

@ -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]]:
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 []
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)
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)]

2
selfdrive/test/helpers.py
View File

@ -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)]

4
selfdrive/test/process_replay/capture.py
View File

@ -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

3
selfdrive/test/process_replay/compare_logs.py
View File

@ -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"

87
selfdrive/test/process_replay/process_replay.py
View File

@ -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]
subs: List[str]
ignore: List[str]
config_callback: Optional[Callable] = None
init_callback: Optional[Callable] = None
should_recv_callback: Optional[Callable] = None
tolerance: Optional[float] = None
pubs: list[str]
subs: list[str]
ignore: list[str]
config_callback: Callable | None = None
init_callback: Callable | None = None
should_recv_callback: Callable | None = 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)
ignore_alive_pubs: List[str] = field(default_factory=list)
unlocked_pubs: List[str] = field(default_factory=list)
vision_pubs: list[str] = field(default_factory=list)
ignore_alive_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.sockets: Optional[List[messaging.SubSocket]] = None
self.rc: Optional[ReplayContext] = None
self.vipc_server: Optional[VisionIpcServer] = None
self.environ_config: Optional[Dict[str, Any]] = None
self.capture: Optional[ProcessOutputCapture] = None
self.pm: messaging.PubMaster | None = None
self.sockets: list[messaging.SubSocket] | None = None
self.rc: ReplayContext | None = None
self.vipc_server: VisionIpcServer | None = None
self.environ_config: dict[str, Any] | None = 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],
all_msgs: LogIterable, frs: Optional[Dict[str, BaseFrameReader]],
fingerprint: Optional[str], capture_output: bool
self, params_config: dict[str, Any], environ_config: dict[str, Any],
all_msgs: LogIterable, frs: dict[str, BaseFrameReader] | None,
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,
fingerprint: Optional[str] = None, return_all_logs: bool = False, custom_params: Optional[Dict[str, Any]] = None,
captured_output_store: Optional[Dict[str, Dict[str, str]]] = None, disable_progress: bool = False
) -> List[capnp._DynamicStructReader]:
cfg: ProcessConfig | Iterable[ProcessConfig], lr: LogIterable, frs: dict[str, BaseFrameReader] | None = None,
fingerprint: str | None = None, return_all_logs: bool = False, custom_params: dict[str, Any] | None = None,
captured_output_store: dict[str, dict[str, str]] | None = None, disable_progress: bool = False
) -> 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],
custom_params: Optional[Dict[str, Any]], captured_output_store: Optional[Dict[str, Dict[str, str]]], disable_progress: bool
) -> List[capnp._DynamicStructReader]:
cfgs: list[ProcessConfig], lr: LogIterable, frs: dict[str, BaseFrameReader] | None, fingerprint: str | None,
custom_params: dict[str, Any] | None, captured_output_store: dict[str, dict[str, str]] | None, disable_progress: bool
) -> 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()
internal_pub_queue: List[capnp._DynamicStructReader] = []
external_pub_queue: list[capnp._DynamicStructReader] = pub_msgs.copy()
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"

11
selfdrive/test/process_replay/regen.py
View File

@ -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__"):

2
selfdrive/test/process_replay/test_debayer.py
View File

@ -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()

6
selfdrive/test/process_replay/test_processes.py
View File

@ -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

6
selfdrive/test/update_ci_routes.py
View File

@ -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']

5
selfdrive/thermald/power_monitoring.py
View File

@ -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

19
selfdrive/thermald/thermald.py
View File

@ -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_prev: Dict[str, bool] = {}
startup_conditions: dict[str, bool] = {}
startup_conditions_prev: dict[str, bool] = {}
off_ts: Optional[float] = None
started_ts: Optional[float] = None
off_ts: float | None = 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(

5
selfdrive/ui/soundd.py
View File

@ -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:

6
selfdrive/ui/tests/test_translations.py
View File

@ -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}")

4
selfdrive/ui/translations/create_badges.py
View File

@ -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

2
selfdrive/ui/update_translations.py
View File

@ -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:

11
selfdrive/updated.py
View File

@ -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)

3
system/hardware/base.py
View File

@ -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}

6
system/hardware/tici/agnos.py
View File

@ -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:

5
system/hardware/tici/amplifier.py
View File

@ -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):

18
system/hardware/tici/casync.py
View File

@ -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],
sources: List[Tuple[str, ChunkReader, ChunkDict]],
def extract(target: list[Chunk],
sources: list[tuple[str, ChunkReader, ChunkDict]],
out_path: str,
progress: Optional[Callable[[int], None]] = None):
stats: Dict[str, int] = defaultdict(int)
progress: Callable[[int], None] | None = None):
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():

3
system/hardware/tici/power_monitor.py
View File

@ -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)

2
system/hardware/tici/precise_power_measure.py
View File

@ -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}")

3
system/hardware/tici/tests/compare_casync_manifest.py
View File

@ -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': [],

7
system/hardware/tici/tests/test_power_draw.py
View File

@ -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

3
system/loggerd/deleter.py
View File

@ -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:

5
system/loggerd/tests/loggerd_tests_common.py
View File

@ -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)

2
system/loggerd/tests/test_deleter.py
View File

@ -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

3
system/loggerd/tests/test_loggerd.py
View File

@ -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()

5
system/loggerd/tests/test_uploader.py
View File

@ -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]

15
system/loggerd/uploader.py
View File

@ -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()

5
system/loggerd/xattr_cache.py
View File

@ -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:

2
system/qcomgpsd/nmeaport.py
View File

@ -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:

6
system/qcomgpsd/qcomgpsd.py
View File

@ -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

Some files were not shown because too many files have changed in this diff Show More