more ty, part 2 (#38379)

2026-07-23 16:32:04 +08:00 · 2026-07-19 11:02:00 -07:00
parent 19ecc37de8
commit f0d93eb32d
57 changed files with 194 additions and 165 deletions
@@ -27,6 +27,8 @@ class Api:
    return api_get(endpoint, method=method, timeout=timeout, access_token=access_token, **params)

  def get_token(self, payload_extra=None, expiry_hours=1):
+    if self.private_key is None:
+      raise RuntimeError("private key is not configured")
    now = datetime.now(UTC).replace(tzinfo=None)
    payload = {
      'identity': self.dongle_id,
@@ -19,7 +19,7 @@ class StreamingDecompressor:
  def __init__(self, url: str) -> None:
    self.buf = b""

-    self.req = requests.get(url, stream=True, headers={'Accept-Encoding': None}, timeout=60)
+    self.req = requests.get(url, stream=True, headers={'Accept-Encoding': 'identity'}, timeout=60)
    self.it = self.req.iter_content(chunk_size=1024 * 1024)
    self.decompressor = lzma.LZMADecompressor(format=lzma.FORMAT_AUTO)
    self.eof = False
@@ -339,7 +339,7 @@ class Tici(HardwareBase):

    # Ensure fan gpio is enabled so fan runs until shutdown, also turned on at boot by the ABL
    gpio_init(GPIO.SOM_ST_IO, True)
-    gpio_set(GPIO.SOM_ST_IO, 1)
+    gpio_set(GPIO.SOM_ST_IO, True)

    # *** IRQ config ***

@@ -389,21 +389,21 @@ class Tici(HardwareBase):
    gpio_init(GPIO.STM_RST_N, True)
    gpio_init(GPIO.STM_BOOT0, True)

-    gpio_set(GPIO.STM_RST_N, 1)
-    gpio_set(GPIO.STM_BOOT0, 0)
+    gpio_set(GPIO.STM_RST_N, True)
+    gpio_set(GPIO.STM_BOOT0, False)
    time.sleep(0.01)
-    gpio_set(GPIO.STM_RST_N, 0)
+    gpio_set(GPIO.STM_RST_N, False)

  def recover_internal_panda(self):
    gpio_init(GPIO.STM_RST_N, True)
    gpio_init(GPIO.STM_BOOT0, True)

-    gpio_set(GPIO.STM_RST_N, 1)
-    gpio_set(GPIO.STM_BOOT0, 1)
+    gpio_set(GPIO.STM_RST_N, True)
+    gpio_set(GPIO.STM_BOOT0, True)
    time.sleep(0.01)
-    gpio_set(GPIO.STM_RST_N, 0)
+    gpio_set(GPIO.STM_RST_N, False)
    time.sleep(0.01)
-    gpio_set(GPIO.STM_BOOT0, 0)
+    gpio_set(GPIO.STM_BOOT0, False)

  def booted(self):
    # this normally boots within 8s, but on rare occasions takes 30+s
@@ -52,7 +52,7 @@ PPPD_CMD = [
  "novj", "novjccomp", "ipcp-accept-local", "ipcp-accept-remote", "nomagic",
  "user", '""', "password", '""',
 ]
-INITIAL_STATE = {
+INITIAL_STATE: dict[str, object] = {
  "seconds_since_boot": 0,
  "state": "INITIALIZING",
  "connected": False, "ip_address": "",
@@ -1,11 +1,13 @@
 import numpy as np
-from numbers import Number
+from collections.abc import Sequence
+
+Gain = int | float | tuple[Sequence[float], Sequence[float]] | list[list[float]]

 class PIDController:
-  def __init__(self, k_p, k_i, k_d=0., pos_limit=1e308, neg_limit=-1e308, rate=100):
-    self._k_p: list[list[float]] = [[0], [k_p]] if isinstance(k_p, Number) else k_p
-    self._k_i: list[list[float]] = [[0], [k_i]] if isinstance(k_i, Number) else k_i
-    self._k_d: list[list[float]] = [[0], [k_d]] if isinstance(k_d, Number) else k_d
+  def __init__(self, k_p: Gain, k_i: Gain, k_d: Gain = 0., pos_limit=1e308, neg_limit=-1e308, rate=100):
+    self._k_p = ([0], [k_p]) if isinstance(k_p, (int, float)) else k_p
+    self._k_i = ([0], [k_i]) if isinstance(k_i, (int, float)) else k_i
+    self._k_d = ([0], [k_d]) if isinstance(k_d, (int, float)) else k_d

    self.set_limits(pos_limit, neg_limit)

@@ -52,7 +52,7 @@ _ar_ox_config = DeviceCameraConfig(CameraConfig(1928, 1208, 2648.0), _ar_ox_fish
 _os_config = DeviceCameraConfig(CameraConfig(2688 // 2, 1520 // 2, 1522.0 * 3 / 4), _os_fisheye, _os_fisheye)
 _neo_config = DeviceCameraConfig(CameraConfig(1164, 874, 910.0), CameraConfig(816, 612, 650.0), _NoneCameraConfig())

-DEVICE_CAMERAS = {
+DEVICE_CAMERAS: dict[tuple[str, str], DeviceCameraConfig] = {
  # A "device camera" is defined by a device type and sensor

  # sensor type was never set on eon/neo/two
@@ -176,4 +176,3 @@ def img_from_device(pt_device):

  pt_img = pt_view/pt_view[:, 2:3]
  return pt_img.reshape(input_shape)[:, :2]
-
@@ -21,6 +21,7 @@ from openpilot.selfdrive.pandad import can_capnp_to_list
 from openpilot.selfdrive.test.helpers import read_segment_list
 from openpilot.common.hardware.hw import DEFAULT_DOWNLOAD_CACHE_ROOT
 from openpilot.tools.lib.logreader import LogReader, LogsUnavailable, openpilotci_source, internal_source, comma_api_source
+from openpilot.tools.lib.file_sources import Source
 from openpilot.tools.lib.route import SegmentName

 SafetyModel = car.CarParams.SafetyModel
@@ -130,7 +131,7 @@ class TestCarModelBase(unittest.TestCase):
      segment_range = f"{cls.test_route.route}/{seg}"

      try:
-        sources = [internal_source] if len(INTERNAL_SEG_LIST) else [openpilotci_source, comma_api_source]
+        sources: list[Source] = [internal_source] if len(INTERNAL_SEG_LIST) else [openpilotci_source, comma_api_source]
        lr = LogReader(segment_range, sources=sources, sort_by_time=True)
        return cls.get_testing_data_from_logreader(lr)
      except (LogsUnavailable, AssertionError):
@@ -1,4 +1,5 @@
 import numpy as np
+from collections.abc import Sequence
 from typing import Any
 from functools import cache

@@ -68,7 +69,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: Sequence[float], min_points_total: int, points_per_bucket: int, rowsize: int) -> None:
    self._rng = np.random.default_rng()
    self.x_bounds = x_bounds
    self.buckets = {bounds: NPQueue(maxlen=points_per_bucket, rowsize=rowsize) for bounds in x_bounds}
@@ -101,7 +102,7 @@ class PointBuckets:
      return points
    return points[self._rng.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: Sequence[Sequence[float]]) -> None:
    for point in points:
      self.add_point(*point)

@@ -66,7 +66,7 @@ class LocationEstimator:
    self.observations = {kind: np.zeros(3, dtype=np.float32) for kind in obs_kinds}
    self.observation_errors = {kind: np.zeros(3, dtype=np.float32) for kind in obs_kinds}

-  def reset(self, t: float, x_initial: np.ndarray = PoseKalman.initial_x, P_initial: np.ndarray = PoseKalman.initial_P):
+  def reset(self, t: float | None, x_initial: np.ndarray = PoseKalman.initial_x, P_initial: np.ndarray = PoseKalman.initial_P):
    self.kf.init_state(x_initial, covs=P_initial, filter_time=t)

  def _validate_sensor_source(self, source: log.SensorEventData.SensorSource):
@@ -57,14 +57,14 @@ class TorqueEstimator(ParameterEstimator):
    self.lag = 0.0
    self.track_all_points = track_all_points  # for offline analysis, without max lateral accel or max steer torque filters
    if decimated:
-      self.min_bucket_points = MIN_BUCKET_POINTS / 10
+      self.min_bucket_points: list[float] = (MIN_BUCKET_POINTS / 10).tolist()
      self.min_points_total = MIN_POINTS_TOTAL_QLOG
      self.fit_points = FIT_POINTS_TOTAL_QLOG
      self.factor_sanity = FACTOR_SANITY_QLOG
      self.friction_sanity = FRICTION_SANITY_QLOG

    else:
-      self.min_bucket_points = MIN_BUCKET_POINTS
+      self.min_bucket_points = MIN_BUCKET_POINTS.tolist()
      self.min_points_total = MIN_POINTS_TOTAL
      self.fit_points = FIT_POINTS_TOTAL
      self.factor_sanity = FACTOR_SANITY
@@ -112,9 +112,10 @@ class TorqueEstimator(ParameterEstimator):
              'latAccelOffset': cache_ltp.latAccelOffsetFiltered,
              'frictionCoefficient': cache_ltp.frictionCoefficientFiltered
            }
-          initial_params['points'] = cache_ltp.points
+          cached_points: list[list[float]] = [list(point) for point in cache_ltp.points]
+          initial_params['points'] = cached_points
          self.decay = cache_ltp.decay
-          self.filtered_points.load_points(initial_params['points'])
+          self.filtered_points.load_points(cached_points)
          cloudlog.info("restored torque params from cache")
      except Exception:
        cloudlog.exception("failed to restore cached torque params")
@@ -89,7 +89,7 @@ class ModelState:
    self.frame_skip = ModelConstants.MODEL_RUN_FREQ // ModelConstants.MODEL_CONTEXT_FREQ
    self.input_queues, self.npy = make_input_queues(self.input_shapes, self.frame_skip, device=self.QUEUE_DEV)
    self.full_frames: dict[str, Tensor] = {}
-    self._blob_cache: dict[int, Tensor] = {}
+    self._blob_cache: dict[tuple[str, int], Tensor] = {}
    self.parser = Parser()
    self.frame_buf_params = {k: get_nv12_info(cam_w, cam_h) for k in ('img', 'big_img')}
    self.run_policy = jits['run_policy']
@@ -41,7 +41,7 @@ class Parser:
    raw = outs[name]
    outs[name] = sigmoid(raw)

-  def parse_mdn(self, name, outs, in_N=0, out_N=1, out_shape=None):
+  def parse_mdn(self, name, outs, in_N=0, out_N=1, out_shape=()):
    if self.check_missing(outs, name):
      return
    raw = outs[name]
@@ -60,7 +60,7 @@ class TestPandad:

  def test_in_reset(self):
    gpio_init(GPIO.STM_RST_N, True)
-    gpio_set(GPIO.STM_RST_N, 1)
+    gpio_set(GPIO.STM_RST_N, True)
    assert not Panda.list()
    self._run_test()

@@ -13,11 +13,11 @@ ALL_STATES = tuple(State.schema.enumerants.values())
 ENABLE_EVENT_TYPES = (ET.ENABLE, ET.PRE_ENABLE, ET.OVERRIDE_LATERAL, ET.OVERRIDE_LONGITUDINAL)


-def make_event(event_types):
-  event = {}
+def make_event(event_types: list[str | None]):
+  EVENTS[0] = {}
  for ev in event_types:
-    event[ev] = NormalPermanentAlert("alert")
-  EVENTS[0] = event
+    if ev is not None:
+      EVENTS[0][ev] = NormalPermanentAlert("alert")
  return 0


@@ -215,7 +215,7 @@ class ProcessContainer:

  def _start_process(self):
    if self.capture is not None:
-      self.process.launcher = LauncherWithCapture(self.capture, self.process.launcher)
+      self.process.launcher = LauncherWithCapture(self.capture, self.process.launcher)  # ty: ignore[invalid-assignment]  # intentional wrapper
    self.process.prepare()
    self.process.start()

@@ -631,10 +631,10 @@ def replay_process(
  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:
+  if isinstance(cfg, ProcessConfig):
    cfgs = [cfg]
+  else:
+    cfgs = list(cfg)

  all_msgs = migrate_all(lr,
                         manager_states=True,
@@ -204,7 +204,10 @@ class FaceAnimator:
      frames_back = round(rewind_elapsed / self._animation.frame_duration)
      frame_index = self._rewind_from - frames_back
      if frame_index <= 0:
-        return self._switch_to_next(now)
+        if self._next is None:
+          self._rewinding = False
+          return self._animation.frames[0]
+        return self._switch_to_next(now, self._next)
      return self._animation.frames[frame_index]

    # Play starting frames first (once)
@@ -223,7 +226,7 @@ class FaceAnimator:

    if self._next is not None:
      if frame_index == 0 and (len(self._animation.frames) == 1 or self._seen_nonzero):
-        return self._switch_to_next(now)
+        return self._switch_to_next(now, self._next)
      # No natural return to frame 0 — start rewinding
      if self._animation.mode in (AnimationMode.ONCE_FORWARD, AnimationMode.REPEAT_FORWARD):
        self._rewinding = True
@@ -232,8 +235,8 @@ class FaceAnimator:

    return self._animation.frames[frame_index]

-  def _switch_to_next(self, now: float) -> list[tuple[int, int]]:
-    self._animation = self._next
+  def _switch_to_next(self, now: float, animation: Animation) -> list[tuple[int, int]]:
+    self._animation = animation
    self._next = None
    self._rewinding = False
    self._seen_nonzero = False
@@ -31,7 +31,11 @@ class MainLayout(Widget):
    # Initialize layouts
    self._home_layout = HomeLayout()
    self._home_body_layout = BodyLayout()
-    self._layouts = {MainState.HOME: self._home_layout, MainState.SETTINGS: SettingsLayout(), MainState.ONROAD: AugmentedRoadView()}
+    self._layouts: dict[MainState, Widget] = {
+      MainState.HOME: self._home_layout,
+      MainState.SETTINGS: SettingsLayout(),
+      MainState.ONROAD: AugmentedRoadView(),
+    }

    self._sidebar_rect = rl.Rectangle(0, 0, 0, 0)
    self._content_rect = rl.Rectangle(0, 0, 0, 0)
@@ -65,7 +65,7 @@ class MetricData:
 class Sidebar(Widget):
  def __init__(self):
    super().__init__()
-    self._net_type = NETWORK_TYPES.get(NetworkType.none)
+    self._net_type = NETWORK_TYPES[NetworkType.none]
    self._net_strength = 0

    self._temp_status = MetricData(tr_noop("TEMP"), tr_noop("GOOD"), Colors.GOOD)
@@ -251,7 +251,7 @@ class MiciHomeLayout(Widget):
    self._egpu_icon.set_visible(ui_state.usbgpu and ui_state.usbgpu_compiled)
    self._egpu_icon_gray.set_visible(ui_state.usbgpu and not ui_state.usbgpu_compiled)
    self._mic_icon.set_visible(ui_state.recording_audio)
-    self._body_icon.set_visible(ui_state.is_body)
+    self._body_icon.set_visible(bool(ui_state.is_body))

    footer_rect = rl.Rectangle(self.rect.x + HOME_PADDING, self.rect.y + self.rect.height - 48, self.rect.width - HOME_PADDING, 48)
    self._status_bar_layout.render(footer_rect)
@@ -146,4 +146,4 @@ class MiciMainLayout(Scroller):

  def _on_body_changed(self):
    self._car_onroad_layout.set_visible(not ui_state.is_body)
-    self._body_onroad_layout.set_visible(ui_state.is_body)
+    self._body_onroad_layout.set_visible(bool(ui_state.is_body))
@@ -16,7 +16,7 @@ from openpilot.system.ui.lib.multilang import tr
 from openpilot.system.ui.widgets import Widget
 from openpilot.selfdrive.ui.ui_state import device, ui_state
 from openpilot.system.ui.widgets.label import UnifiedLabel
-from openpilot.system.ui.widgets.html_render import HtmlModal, HtmlRenderer
+from openpilot.system.ui.widgets.html_render import HtmlRenderer
 from openpilot.system.athena.registration import UNREGISTERED_DONGLE_ID


@@ -160,7 +160,7 @@ class DeviceLayoutMici(NavScroller):
  def __init__(self):
    super().__init__()

-    self._fcc_dialog: HtmlModal | None = None
+    self._fcc_dialog: MiciFccModal | None = None

    def power_off_callback():
      ui_state.params.put_bool("DoShutdown", True, block=True)
@@ -219,7 +219,7 @@ class CameraView(Widget):
      [0.0, 0.0, 1.0]
    ])

-  def _render(self, rect: rl.Rectangle):
+  def _render(self, rect: rl.Rectangle, /):
    if self._switching:
      self._handle_switch()

@@ -44,8 +44,8 @@ class ModelPoints:

@dataclass
 class LeadVehicle:
-  glow: list[float] = field(default_factory=list)
-  chevron: list[float] = field(default_factory=list)
+  glow: list[tuple[float, float]] = field(default_factory=list)
+  chevron: list[tuple[float, float]] = field(default_factory=list)
  fill_alpha: int = 0


@@ -37,8 +37,8 @@ class ModelPoints:

@dataclass
 class LeadVehicle:
-  glow: list[float] = field(default_factory=list)
-  chevron: list[float] = field(default_factory=list)
+  glow: list[tuple[float, float]] = field(default_factory=list)
+  chevron: list[tuple[float, float]] = field(default_factory=list)
  fill_alpha: int = 0


@@ -357,7 +357,7 @@ def build_mici_script(pm: PubMaster, main_layout, script: Script) -> None:
    params = Params()
    main_layout._alerts_layout._pending_params = ({"UpdaterNewDescription": params.get("UpdaterNewDescription")} |
                                                  {alert_data.key: params.get(alert_data.key) for alert_data in main_layout._alerts_layout.sorted_alerts})
-    main_layout._alerts_layout._refresh()
+    main_layout._alerts_layout._update_state()

  swipe_right(width, wait_after=WAIT_SHORT)  # open alerts
  script.setup(setup_offroad_alerts_and_refresh)  # show alerts
@@ -49,7 +49,7 @@ def patch_submaster(message_chunks):
        sm.recv_frame[service] = sm.frame
        sm.valid[service] = True
    sm.frame += 1
-  ui_state.sm.update = mock_update
+  ui_state.sm.update = mock_update  # ty: ignore[invalid-assignment]  # profiling hook


 if __name__ == "__main__":
@@ -67,22 +67,22 @@ def parse_po(path: str | Path) -> tuple[POEntry | None, list[POEntry]]:
  cur_field: str | None = None
  plural_idx = 0

-  def finish():
-    nonlocal cur, header
-    if cur is None:
+  def finish(entry: POEntry | None):
+    nonlocal header
+    if entry is None:
      return
-    if cur.msgid == "" and cur.msgstr:
-      header = cur
-    elif cur.msgid != "" or cur.is_plural:
-      entries.append(cur)
-    cur = None
+    if entry.msgid == "" and entry.msgstr:
+      header = entry
+    elif entry.msgid != "" or entry.is_plural:
+      entries.append(entry)

  for raw in lines:
    line = raw.rstrip('\n')
    stripped = line.strip()

    if not stripped:
-      finish()
+      finish(cur)
+      cur = None
      cur_field = None
      continue

@@ -123,6 +123,8 @@ def parse_po(path: str | Path) -> tuple[POEntry | None, list[POEntry]]:
      continue

    if stripped.startswith('msgstr '):
+      if cur is None:
+        cur = POEntry()
      cur.msgstr = _parse_quoted(stripped[len('msgstr '):])
      cur_field = 'msgstr'
      continue
@@ -138,7 +140,7 @@ def parse_po(path: str | Path) -> tuple[POEntry | None, list[POEntry]]:
      elif cur_field == 'msgstr_plural':
        cur.msgstr_plural[plural_idx] += val

-  finish()
+  finish(cur)
  return header, entries


@@ -669,7 +669,10 @@ def log_handler(end_event: threading.Event) -> None:
 def ws_proxy_recv(ws: WebSocket, local_sock: socket.socket, ssock: socket.socket, end_event: threading.Event, global_end_event: threading.Event) -> None:
  while not (end_event.is_set() or global_end_event.is_set()):
    try:
-      r = select.select((ws.sock,), (), (), 30)
+      sock = ws.sock
+      if sock is None:
+        return
+      r = select.select((sock,), (), (), 30)
      if r[0]:
        data = ws.recv()
        if isinstance(data, str):
@@ -60,7 +60,7 @@ class TestAthenadMethods:
  @classmethod
  def setup_class(cls):
    cls.SOCKET_PORT = 45454
-    athenad.Api = MockApi
+    athenad.Api = MockApi  # ty: ignore[invalid-assignment]  # test double
    athenad.LOCAL_PORT_WHITELIST = {cls.SOCKET_PORT}

  def setup_method(self):
@@ -351,6 +351,7 @@ class TestAthenadMethods:
    assert items[0] == asdict(item)
    assert not items[0]['current']

+    assert item.id is not None
    athenad.cancelled_uploads.add(item.id)
    items = dispatcher["listUploadQueue"]()
    assert len(items) == 0
@@ -363,6 +364,7 @@ class TestAthenadMethods:
    athenad.upload_queue.put_nowait(item2)

    # Ensure canceled items are not persisted
+    assert item2.id is not None
    athenad.cancelled_uploads.add(item2.id)

    # serialize item
@@ -34,7 +34,7 @@ class PowerMonitoring:
    self.car_battery_capacity_uWh = max((CAR_BATTERY_CAPACITY_uWh / 10), car_battery_capacity_uWh)

  # Calculation tick
-  def calculate(self, voltage: int | None, ignition: bool):
+  def calculate(self, voltage: float | None, ignition: bool):
    try:
      now = time.monotonic()

@@ -35,7 +35,7 @@ class TestPowerMonitoring:
  def test_panda_state_present(self):
    pm = PowerMonitoring()
    for _ in range(10):
-      pm.calculate(None, None)
+      pm.calculate(None, False)
    assert pm.get_power_used() == 0
    assert pm.get_car_battery_capacity() == (CAR_BATTERY_CAPACITY_uWh / 10)

@@ -63,10 +63,10 @@ class UploaderTestCase:
  seg_dir: str

  def set_ignore(self):
-    uploader.Api = MockApiIgnore
+    uploader.Api = MockApiIgnore  # ty: ignore[invalid-assignment]  # test double

  def setup_method(self):
-    uploader.Api = MockApi
+    uploader.Api = MockApi  # ty: ignore[invalid-assignment]  # test double
    uploader.fake_upload = True
    uploader.force_wifi = True
    uploader.allow_sleep = False
@@ -19,7 +19,7 @@ class TestDeleter(UploaderTestCase):
    self.f_type = "fcamera.hevc"
    super().setup_method()
    self.fake_stats = Stats(f_bavail=0, f_blocks=10, f_frsize=4096)
-    deleter.os.statvfs = self.fake_statvfs
+    deleter.os.statvfs = self.fake_statvfs  # ty: ignore[invalid-assignment]  # test double

  def start_thread(self):
    self.end_event = threading.Event()
@@ -5,6 +5,7 @@ import random
 import string
 import subprocess
 import time
+from collections.abc import Collection
 from collections import defaultdict
 from pathlib import Path
 import pytest
@@ -74,8 +75,8 @@ 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]:
-    pm = messaging.PubMaster(services)
+  def _publish_random_messages(self, services: Collection[str]) -> dict[str, list]:
+    pm = messaging.PubMaster(list(services))

    managed_processes["loggerd"].start()
    for s in services:
@@ -1,9 +1,9 @@
 import os
 import sys
-from dataclasses import dataclass, fields
+from dataclasses import dataclass
 from subprocess import check_output, CalledProcessError
 from time import sleep
-from typing import NoReturn, cast
+from typing import NoReturn

 DEBUG = int(os.environ.get("DEBUG", "0"))

@@ -17,27 +17,27 @@ class GnssClockNmeaPort:
  # 0x10 = bias_uncertainty_ns valid
  # 0x20 = drift_nsps valid
  # 0x40 = drift_uncertainty_nsps valid
-  flags: int
-  leap_seconds: int
-  time_ns: int
-  time_uncertainty_ns: int # 1-sigma
-  full_bias_ns: int
-  bias_ns: float
-  bias_uncertainty_ns: float # 1-sigma
-  drift_nsps: float
-  drift_uncertainty_nsps: float # 1-sigma
+  flags: int | None
+  leap_seconds: int | None
+  time_ns: int | None
+  time_uncertainty_ns: int | None # 1-sigma
+  full_bias_ns: int | None
+  bias_ns: float | None
+  bias_uncertainty_ns: float | None # 1-sigma
+  drift_nsps: float | None
+  drift_uncertainty_nsps: float | None # 1-sigma

-  def __post_init__(self):
-    for field in fields(self):
-      val = getattr(self, field.name)
-      field_type = cast(type, field.type)
-      setattr(self, field.name, field_type(val) if val else None)
+  @classmethod
+  def from_fields(cls, values: list[str]) -> 'GnssClockNmeaPort':
+    ints = [int(value) if value else None for value in values[:5]]
+    floats = [float(value) if value else None for value in values[5:9]]
+    return cls(*ints, *floats)

@dataclass
 class GnssMeasNmeaPort:
-  messageCount: int
-  messageNum: int
-  svCount: int
+  messageCount: int | None
+  messageNum: int | None
+  svCount: int | None
  # constellation enum:
  # 1 = GPS
  # 2 = SBAS
@@ -45,10 +45,10 @@ class GnssMeasNmeaPort:
  # 4 = QZSS
  # 5 = BEIDOU
  # 6 = GALILEO
-  constellation: int
-  svId: int
-  flags: int # always zero
-  time_offset_ns: int
+  constellation: int | None
+  svId: int | None
+  flags: int | None # always zero
+  time_offset_ns: int | None
  # state bit mask:
  # 0x0001 = CODE LOCK
  # 0x0002 = BIT SYNC
@@ -64,18 +64,18 @@ class GnssMeasNmeaPort:
  # 0x0800 = GALILEO E1C 2ND CODE LOCK
  # 0x1000 = GALILEO E1B PAGE SYNC
  # 0x2000 = GALILEO E1B PAGE SYNC
-  state: int
-  time_of_week_ns: int
-  time_of_week_uncertainty_ns: int # 1-sigma
-  carrier_to_noise_ratio: float
-  pseudorange_rate: float
-  pseudorange_rate_uncertainty: float # 1-sigma
+  state: int | None
+  time_of_week_ns: int | None
+  time_of_week_uncertainty_ns: int | None # 1-sigma
+  carrier_to_noise_ratio: float | None
+  pseudorange_rate: float | None
+  pseudorange_rate_uncertainty: float | None # 1-sigma

-  def __post_init__(self):
-    for field in fields(self):
-      val = getattr(self, field.name)
-      field_type = cast(type, field.type)
-      setattr(self, field.name, field_type(val) if val else None)
+  @classmethod
+  def from_fields(cls, values: list[str]) -> 'GnssMeasNmeaPort':
+    ints = [int(value) if value else None for value in values[:10]]
+    floats = [float(value) if value else None for value in values[10:13]]
+    return cls(*ints, *floats)

 def nmea_checksum_ok(s):
  checksum = 0
@@ -109,11 +109,11 @@ def process_nmea_port_messages(device:str="/dev/ttyUSB1") -> NoReturn:
          match fields[0]:
            case "$GNCLK":
              # fields at end are reserved (not used)
-              gnss_clock = GnssClockNmeaPort(*fields[1:10])
+              gnss_clock = GnssClockNmeaPort.from_fields(fields[1:10])
              print(gnss_clock)
            case "$GNMEAS":
              # fields at end are reserved (not used)
-              gnss_meas = GnssMeasNmeaPort(*fields[1:14])
+              gnss_meas = GnssMeasNmeaPort.from_fields(fields[1:14])
              print(gnss_meas)
    except Exception as e:
      print(e)
@@ -20,7 +20,7 @@ SENSOR_CONFIGS = (
 )
 SENSOR_CONFIGS_BY_MEASUREMENT = {config.measurement: config for config in SENSOR_CONFIGS}

-def get_irq_count(irq: int):
+def get_irq_count(irq: str):
  with open(f"/sys/kernel/irq/{irq}/per_cpu_count") as f:
    per_cpu = map(int, f.read().split(","))
    return sum(per_cpu)
@@ -184,7 +184,7 @@ class BinaryStruct:
        setattr(obj, name, value)
      return obj

-    cls._read = _read
+    cls._read = _read  # ty: ignore[invalid-assignment]  # installed dynamically for each subclass

  @classmethod
  def from_bytes(cls: type[T], data: bytes) -> T:
@@ -1,8 +1,9 @@
 import pyray as rl
+from collections.abc import Sequence


 class GuiStyleContext:
-  def __init__(self, styles: list[tuple[int, int, int]]):
+  def __init__(self, styles: Sequence[tuple[int, int, int]]):
    """styles is a list of tuples (control, prop, new_value)"""
    self.styles = styles
    self.prev_styles: list[tuple[int, int, int]] = []
@@ -257,7 +257,7 @@ class WifiManager:

  def add_callbacks(self, need_auth: Callable[[str], None] | None = None,
                    activated: Callable[[], None] | None = None,
-                    forgotten: Callable[[str], None] | None = None,
+                    forgotten: Callable[[str | None], None] | None = None,
                    networks_updated: Callable[[list[Network]], None] | None = None,
                    disconnected: Callable[[], None] | None = None):
    if need_auth is not None:
@@ -37,7 +37,11 @@ class Reset(Widget):
    self._reset_state = ResetState.NONE
    self._cancel_button = Button("Cancel", gui_app.request_close)
    self._confirm_button = Button("Confirm", self._confirm, button_style=ButtonStyle.PRIMARY)
-    self._reboot_button = Button("Reboot", lambda: subprocess.run("sudo reboot", shell=True))
+    self._reboot_button = Button("Reboot", self._reboot)
+
+  @staticmethod
+  def _reboot() -> None:
+    subprocess.run("sudo reboot", shell=True)

  def _do_erase(self):
    if PC:
@@ -3,16 +3,25 @@ from __future__ import annotations
 import abc
 import pyray as rl
 from enum import IntEnum
-from typing import TypeVar
+from typing import Protocol, TypeVar
 from collections.abc import Callable
 from openpilot.system.ui.lib.application import gui_app, MousePos, MAX_TOUCH_SLOTS, MouseEvent

-try:
-  from openpilot.selfdrive.ui.ui_state import device
-except ImportError:
-  class Device:
-    awake = True
-  device = Device()
+class DeviceLike(Protocol):
+  awake: bool
+
+
+def _get_device() -> DeviceLike:
+  try:
+    from openpilot.selfdrive.ui.ui_state import device
+    return device
+  except ImportError:
+    class Device:
+      awake = True
+    return Device()
+
+
+device = _get_device()

 W = TypeVar('W', bound='Widget')

@@ -185,16 +194,16 @@ class Widget(abc.ABC):
    """Optionally update the widget's non-layout state. This is called before rendering."""

  @abc.abstractmethod
-  def _render(self, rect: rl.Rectangle) -> bool | int | None:
+  def _render(self, rect: rl.Rectangle, /) -> bool | int | None:
    """Render the widget within the given rectangle."""

  def _update_layout_rects(self) -> None:
    """Optionally update any layout rects on Widget rect change."""

-  def _handle_mouse_press(self, mouse_pos: MousePos) -> None:
+  def _handle_mouse_press(self, mouse_pos: MousePos, /) -> None:
    """Optionally handle mouse press events."""

-  def _handle_mouse_release(self, mouse_pos: MousePos) -> None:
+  def _handle_mouse_release(self, mouse_pos: MousePos, /) -> None:
    """Optionally handle mouse release events."""
    if self._click_delay is not None:
      self._click_release_time = rl.get_time() + self._click_delay
@@ -1,7 +1,6 @@
 import math
 from enum import IntEnum
 from collections.abc import Callable
-from itertools import zip_longest
 from typing import Union
 import pyray as rl

@@ -210,7 +209,7 @@ class Label(Widget):
        icon_x = self._rect.x + (self._rect.width - self._icon.width) / 2
      rl.draw_texture_v(self._icon, rl.Vector2(icon_x, icon_y), rl.WHITE)

-    for text, text_size, emojis in zip_longest(self._text_wrapped, self._text_size, self._emojis, fillvalue=[]):
+    for text, text_size, emojis in zip(self._text_wrapped, self._text_size, self._emojis, strict=True):
      line_pos = rl.Vector2(text_pos.x, text_pos.y)
      if self._text_alignment == rl.GuiTextAlignment.TEXT_ALIGN_LEFT:
        line_pos.x += self._text_padding
@@ -78,7 +78,7 @@ class NavWidget(Widget, abc.ABC):
    # the top of a vertical scroll panel to prevent erroneous swipes
    return True

-  def set_back_callback(self, callback: Callable[[], None]) -> None:
+  def set_back_callback(self, callback: Callable[[], None] | None) -> None:
    self._back_callback = callback

  def set_shown_callback(self, callback: Callable[[], None] | None) -> None:
@@ -15,16 +15,6 @@ from openpilot.system.ui.widgets.label import gui_label
 from openpilot.system.ui.widgets.scroller_tici import Scroller
 from openpilot.system.ui.widgets.list_view import ButtonAction, ListItem, MultipleButtonAction, ToggleAction, button_item, text_item

-# These are only used for AdvancedNetworkSettings, standalone apps just need WifiManagerUI
-try:
-  from openpilot.common.params import Params
-  from openpilot.selfdrive.ui.ui_state import ui_state
-  from openpilot.selfdrive.ui.lib.prime_state import PrimeType
-except Exception:
-  Params = None
-  ui_state = None
-  PrimeType = None
-
 NM_DEVICE_STATE_NEED_AUTH = 60
 MIN_PASSWORD_LENGTH = 8
 MAX_PASSWORD_LENGTH = 64
@@ -105,11 +95,16 @@ class NetworkUI(Widget):

 class AdvancedNetworkSettings(Widget):
  def __init__(self, wifi_manager: WifiManager):
-    assert Params is not None
+    # AdvancedNetworkSettings needs the full openpilot environment, standalone apps just use WifiManagerUI
+    from openpilot.common.params import Params
+    from openpilot.selfdrive.ui.ui_state import ui_state
+    from openpilot.selfdrive.ui.lib.prime_state import PrimeType
    super().__init__()
    self._wifi_manager = wifi_manager
    self._wifi_manager.add_callbacks(networks_updated=self._on_network_updated)
    self._params = Params()
+    self._prime_state = ui_state.prime_state
+    self._cell_prime_types = (PrimeType.NONE, PrimeType.LITE)

    self._keyboard = Keyboard(max_text_size=MAX_PASSWORD_LENGTH, min_text_size=MIN_PASSWORD_LENGTH, show_password_toggle=True)

@@ -254,7 +249,7 @@ class AdvancedNetworkSettings(Widget):
    self._wifi_manager.process_callbacks()

    # If not using prime SIM, show GSM settings and enable IPv4 forwarding
-    show_cell_settings = ui_state.prime_state.get_type() in (PrimeType.NONE, PrimeType.LITE)
+    show_cell_settings = self._prime_state.get_type() in self._cell_prime_types
    self._wifi_manager.set_ipv4_forward(show_cell_settings)
    self._roaming_btn.set_visible(show_cell_settings)
    self._apn_btn.set_visible(show_cell_settings)
@@ -1,6 +1,6 @@
 import pyray as rl
 import numpy as np
-from collections.abc import Callable
+from collections.abc import Callable, Sequence

 from openpilot.common.filter_simple import FirstOrderFilter, BounceFilter
 from openpilot.common.swaglog import cloudlog
@@ -40,7 +40,7 @@ class ScrollIndicator(Widget):
    self._content_size = content_size
    self._viewport = viewport

-  def _render(self, _):
+  def _render(self, _, /):
    # scale indicator width based on content size
    indicator_w = float(np.interp(self._content_size, [1000, 3000], [300, 100]))

@@ -69,7 +69,7 @@ class ScrollIndicator(Widget):

 class _Scroller(Widget):
  """Should use wrapper below to reduce boilerplate"""
-  def __init__(self, items: list[Widget], horizontal: bool = True, snap_items: bool = False, spacing: int = ITEM_SPACING,
+  def __init__(self, items: Sequence[Widget], horizontal: bool = True, snap_items: bool = False, spacing: int = ITEM_SPACING,
               pad: int = ITEM_SPACING, scroll_indicator: bool = True, edge_shadows: bool = True):
    super().__init__()
    self._items: list[Widget] = []
@@ -150,7 +150,7 @@ class _Scroller(Widget):
                                          and not self.moving_items and (original_touch_valid_callback() if
                                                                         original_touch_valid_callback else True))

-  def add_widgets(self, items: list[Widget]) -> None:
+  def add_widgets(self, items: Sequence[Widget]) -> None:
    for item in items:
      self.add_widget(item)

@@ -332,7 +332,7 @@ class _Scroller(Widget):
    else:
      item.render()

-  def _render(self, _):
+  def _render(self, _, /):
    rl.begin_scissor_mode(int(self._rect.x), int(self._rect.y),
                          int(self._rect.width), int(self._rect.height))

@@ -397,7 +397,7 @@ class Scroller(Widget):
    # pass down enabled to child widget for nav stack
    self._scroller.set_enabled(lambda: self.enabled)

-  def _render(self, _):
+  def _render(self, _, /):
    self._scroller.render(self._rect)


@@ -1,4 +1,5 @@
 import pyray as rl
+from collections.abc import Sequence
 from openpilot.system.ui.lib.scroll_panel import GuiScrollPanel
 from openpilot.system.ui.widgets import Widget

@@ -23,7 +24,7 @@ class LineSeparator(Widget):


 class Scroller(Widget):
-  def __init__(self, items: list[Widget], spacing: int = ITEM_SPACING, line_separator: bool = False, pad_end: bool = True):
+  def __init__(self, items: Sequence[Widget], spacing: int = ITEM_SPACING, line_separator: bool = False, pad_end: bool = True):
    super().__init__()
    self._items: list[Widget] = []
    self._spacing = spacing
@@ -357,7 +357,7 @@ class Updater:
    setup_git_options(OVERLAY_MERGED)
    output = run(["git", "ls-remote", "--heads"], OVERLAY_MERGED)

-    self.branches = defaultdict(lambda: None)
+    self.branches.clear()
    for line in output.split('\n'):
      ls_remotes_re = r'(?P<commit_sha>\b[0-9a-f]{5,40}\b)(\s+)(refs\/heads\/)(?P<branch_name>.*$)'
      x = re.fullmatch(ls_remotes_re, line.strip())
@@ -55,9 +55,11 @@ class TestStreamSession:

      mocked_pubmaster.send.assert_called_once()
      mt, md = mocked_pubmaster.send.call_args.args
-      assert mt == msg["type"]
+      msg_type = msg["type"]
+      assert isinstance(msg_type, str)
+      assert mt == msg_type
      assert isinstance(md, capnp._DynamicStructBuilder)
-      assert hasattr(md, msg["type"])
+      assert hasattr(md, msg_type)

      mocked_pubmaster.reset_mock()

@@ -519,7 +519,7 @@ class WebrtcdHandler(BaseHTTPRequestHandler):
  def do_OPTIONS(self) -> None:
    self._dispatch_request()

-  def log_message(self, fmt, *args) -> None:
+  def log_message(self, format: str, *args: object) -> None:  # noqa: A002  # stdlib override
    # silence default access logging; errors are logged explicitly in _dispatch_request
    pass

@@ -54,7 +54,7 @@ class ClientRedirectHandler(BaseHTTPRequestHandler):
    self.end_headers()
    self.wfile.write(b'Return to the CLI to continue')

-  def log_message(self, *args):
+  def log_message(self, format: str, *args: object) -> None:  # noqa: A002  # stdlib override
    pass  # this prevent http server from dumping messages to stdout


@@ -12,7 +12,7 @@ Source = Callable[[SegmentRange, list[int], FileNames], dict[int, str]]
 InternalUnavailableException = Exception("Internal source not available")


-def comma_api_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames) -> dict[int, str]:
+def comma_api_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames, /) -> dict[int, str]:
  route = Route(sr.route_name)

  # comma api will have already checked if the file exists
@@ -22,7 +22,7 @@ def comma_api_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames) -> d
    return {seg: route.qlog_paths()[seg] for seg in seg_idxs if route.qlog_paths()[seg] is not None}


-def internal_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames, endpoint_url: str = DATA_ENDPOINT) -> dict[int, str]:
+def internal_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames, /, endpoint_url: str = DATA_ENDPOINT) -> dict[int, str]:
  if not internal_source_available(endpoint_url):
    raise InternalUnavailableException

@@ -32,11 +32,11 @@ def internal_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames, endpo
  return eval_source({seg: [get_internal_url(sr, seg, fn) for fn in fns] for seg in seg_idxs})


-def openpilotci_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames) -> dict[int, str]:
+def openpilotci_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames, /) -> dict[int, str]:
  return eval_source({seg: [get_url(sr.route_name, seg, fn) for fn in fns] for seg in seg_idxs})


-def comma_car_segments_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames) -> dict[int, str]:
+def comma_car_segments_source(sr: SegmentRange, seg_idxs: list[int], fns: FileNames, /) -> dict[int, str]:
  return eval_source({seg: get_comma_segments_url(sr.route_name, seg) for seg in seg_idxs})


@@ -16,7 +16,7 @@ class CachingTestRequestHandler(http.server.BaseHTTPRequestHandler):

  def do_GET(self):
    if self.FILE_EXISTS:
-      self.send_response(206 if "Range" in self.headers else 200, b'1234')
+      self.send_response(206 if "Range" in self.headers else 200, '1234')
    else:
      self.send_response(404)
    self.end_headers()
@@ -59,7 +59,7 @@ class Maneuver:

    return float(action_accel)

-  def get_accel(self, v_ego: float, long_active: bool, standstill: bool, cruise_standstill: bool) -> float:
+  def get_accel(self, v_ego: float, long_active: bool, standstill: bool, cruise_standstill: bool, /) -> float:
    ready = abs(v_ego - self.initial_speed) < 0.3 and long_active and not cruise_standstill
    if self.initial_speed < 0.01:
      ready = ready and standstill
@@ -5,6 +5,7 @@ import numpy as np
 import pyray as rl

 from matplotlib.backends.backend_agg import FigureCanvasAgg
+from matplotlib.artist import Artist
 from matplotlib.offsetbox import AnchoredOffsetbox, HPacker, TextArea

 from openpilot.common.transformations.camera import get_view_frame_from_calib_frame
@@ -119,7 +120,7 @@ def init_plots(arr, name_to_arr_idx, plot_xlims, plot_ylims, plot_names, plot_co
      idxs.append(name_to_arr_idx[item])
      plot_select.append(i)
    # Build colored title: each label colored to match its plot line
-    title_texts = []
+    title_texts: list[Artist] = []
    for j2, (nm, cl) in enumerate(zip(pl_list, plot_colors[i], strict=False)):
      if j2 > 0:
        title_texts.append(TextArea(", ", textprops={"color": "white", "fontsize": 10}))
@@ -94,7 +94,7 @@ Ignition: {self.simulator_state.ignition} Engaged: {self.simulator_state.is_enga
    """)

  @abstractmethod
-  def spawn_world(self, q: Queue) -> World:
+  def spawn_world(self, q: Queue, /) -> World:
    pass

  def _run(self, q: Queue):
@@ -40,7 +40,7 @@ class SimulatorState:
    self.is_engaged = False
    self.ignition = True

-    self.velocity: vec3 = None
+    self.velocity = vec3(0, 0, 0)
    self.bearing: float = 0
    self.gps = GPSState()
    self.imu = IMUState()
@@ -72,7 +72,7 @@ class World(ABC):
    self.exit_event = multiprocessing.Event()

  @abstractmethod
-  def apply_controls(self, steer_sim, throttle_out, brake_out):
+  def apply_controls(self, steer_sim, throttle_out, brake_out, /):
    pass

  @abstractmethod
@@ -84,7 +84,7 @@ class World(ABC):
    pass

  @abstractmethod
-  def read_sensors(self, simulator_state: SimulatorState):
+  def read_sensors(self, simulator_state: SimulatorState, /):
    pass

  @abstractmethod
@@ -186,10 +186,6 @@ quote-style = "preserve"
 [tool.ty.rules]
 unresolved-import = "ignore"  # Cython-compiled modules (.pyx)
 unresolved-attribute = "ignore"  # many from capnp and Cython modules
-invalid-method-override = "ignore"  # signature variance issues
-possibly-missing-attribute = "ignore"  # too many false positives
-invalid-assignment = "ignore"  # often intentional monkey-patching
-invalid-argument-type = "ignore"  # many false positives from raylib, ctypes, numpy

 [tool.uv]
 python-preference = "only-managed"