lint

openpilot/common/api/base.py

@@ -38,6 +38,7 @@ class BaseApi:
     }
     if payload_extra is not None:
       payload.update(payload_extra)
+    assert self.private_key is not None
     token = jwt.encode(payload, self.private_key, algorithm=self.jwt_algorithm)
     if isinstance(token, bytes):
       token = token.decode('utf8')

openpilot/selfdrive/controls/controlsd.py

@@ -147,6 +147,7 @@ class Controls(ControlsExt):
     lat_delay = self.sm["liveDelay"].lateralDelay + LAT_SMOOTH_SECONDS
 
     actuators.curvature = self.desired_curvature
+    assert self.calibrated_pose is not None
     steer, lateral_output, lac_log = self.LaC.update(CC.latActive, CS, self.VM, lp,
                                                      self.steer_limited_by_safety, self.desired_curvature,
                                                      self.calibrated_pose, curvature_limited, lat_delay)

openpilot/selfdrive/monitoring/test_monitoring.py

@@ -271,7 +271,7 @@ def test_run_step_engagement(selfdrive_enabled, lat_active, steering, gas,
     captured['op_engaged'] = op_engaged
     return orig(driver_engaged, op_engaged, lowspeed, wrong_gear)
 
-  dm._update_events = spy
+  object.__setattr__(dm, '_update_events', spy)
   dm.run_step(sm, demo=False)
   assert captured['op_engaged'] == expected_op_engaged
   assert captured['driver_engaged'] == expected_driver_engaged

openpilot/selfdrive/ui/layouts/sidebar.py

@@ -68,7 +68,7 @@ class Sidebar(Widget, SidebarSP):
   def __init__(self):
     Widget.__init__(self)
     SidebarSP.__init__(self)
-    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)

openpilot/selfdrive/ui/sunnypilot/layouts/settings/models.py

@@ -238,7 +238,7 @@ class ModelsLayout(Widget):
     self.lagd_toggle.action_item.set_state(live_delay)
     self.delay_control.set_visible(not live_delay and advanced_controls)
     new_step = int(round(100 / CV.MPH_TO_KPH)) if ui_state.is_metric else 100
-    if self.lane_turn_value_control.action_item.value_change_step != new_step:
+    if self.lane_turn_value_control.action_item is not None and self.lane_turn_value_control.action_item.value_change_step != new_step:
       self.lane_turn_value_control.action_item.value_change_step = new_step
 
     self._update_lagd_description(live_delay)

openpilot/selfdrive/ui/sunnypilot/layouts/settings/network.py

@@ -38,8 +38,8 @@ class NetworkUISP(NetworkUI):
       self.scan_button.set_text(tr("Scan"))
       self.scan_button.set_enabled(True)
 
-  def _render(self, rect: rl.Rectangle):
-    super()._render(rect)
+  def _render(self, _):
+    super()._render(_)
 
     if self._current_panel == PanelType.WIFI:
       self.scan_button.set_position(self._rect.x, self._rect.y + 20)

openpilot/selfdrive/ui/sunnypilot/layouts/settings/settings.py

@@ -37,7 +37,7 @@ from openpilot.system.ui.widgets.scroller_tici import Scroller
 OP.PANEL_COLOR = rl.Color(10, 10, 10, 255)
 ICON_SIZE = 70
 
-OP.PanelType = IntEnum(
+OP.PanelType = IntEnum(  # type: ignore[assignment] # ty: ignore[invalid-assignment]
   "PanelType",
   [es.name for es in OP.PanelType] + [
     "SUNNYLINK",
@@ -180,20 +180,18 @@ class SettingsLayoutSP(OP.SettingsLayout):
       self._sidebar_scroller.render(nav_rect)
       return
 
-  def _handle_mouse_release(self, mouse_pos: MousePos) -> bool:
+  def _handle_mouse_release(self, mouse_pos: MousePos) -> None:
     # Check close button
     if rl.check_collision_point_rec(mouse_pos, self._close_btn_rect):
       if self._close_callback:
         self._close_callback()
-      return True
+      return
 
     # Check navigation buttons
     for panel_type, panel_info in self._panels.items():
       if rl.check_collision_point_rec(mouse_pos, panel_info.button_rect) and self._sidebar_scroller.scroll_panel.is_touch_valid():
         self.set_current_panel(panel_type)
-        return True
-
-    return False
+        return
 
   def show_event(self):
     super().show_event()

openpilot/selfdrive/ui/sunnypilot/mici/layouts/models.py

@@ -138,7 +138,7 @@ class ModelsLayoutMici(NavScroller):
     self._show_selection_view(btns, self._show_folders)
 
   def _reset_main_view(self):
-    self._scroller._items = self.main_items
+    self._scroller._items = self.main_items  # type: ignore[assignment] # ty: ignore[invalid-assignment]
     self.set_back_callback(self.original_back_callback)
     self._scroller.scroll_panel.set_offset(0)
     self._scroller.scroll_to(0)

openpilot/selfdrive/ui/sunnypilot/mici/layouts/onboarding.py

@@ -16,6 +16,9 @@ class SunnylinkConsentPage(NavScroller):
   def __init__(self, on_accept: Callable | None = None, on_decline: Callable | None = None):
     super().__init__()
 
+    assert on_accept is not None and callable(on_accept)
+    assert on_decline is not None and callable(on_decline)
+
     self._accept_button = BigConfirmationCircleButton("enable\nsunnylink", gui_app.texture("icons_mici/setup/driver_monitoring/dm_check.png", 64, 64),
                                                       on_accept,  exit_on_confirm=False)
 

openpilot/selfdrive/ui/sunnypilot/onroad/speed_limit.py

@@ -198,7 +198,7 @@ class SpeedLimitRenderer(Widget, SpeedLimitAlertRenderer):
         self._draw_ahead_info(sign_rect)
 
   def _draw_sign_main(self, rect, alpha=1.0):
-    speed_limit_warning_enabled = ui_state.speed_limit_mode >= SpeedLimitMode.warning
+    speed_limit_warning_enabled = ui_state.speed_limit_mode is not None and ui_state.speed_limit_mode >= SpeedLimitMode.warning
     has_limit = self.speed_limit_valid or self.speed_limit_last_valid
     is_overspeed = has_limit and round(self.speed_limit_final_last) < round(self.speed)
 

openpilot/selfdrive/ui/sunnypilot/ui_state.py

@@ -224,8 +224,7 @@ class UIStateSP:
 
 
 class DeviceSP:
-  @staticmethod
-  def _set_awake(on: bool, _ui_state):
+  def _set_awake(self, on: bool, _ui_state=None):
     if _ui_state.boot_offroad_mode == 1 and not on:
       _ui_state.params.put_bool("OffroadMode", True)
 

openpilot/selfdrive/ui/ui_state.py

@@ -314,9 +314,9 @@ class Device(DeviceSP):
       brightness = 0
 
     if brightness != self._last_brightness:
-      self._brightness_target = brightness
+      self._brightness_target = int(brightness)
       self._brightness_event.set()
-      self._last_brightness = brightness
+      self._last_brightness = int(brightness)
 
   def _update_wakefulness(self):
     # Handle interactive timeout
@@ -337,9 +337,9 @@ class Device(DeviceSP):
 
     self._set_awake(ui_state.ignition or not interaction_timeout or PC)
 
-  def _set_awake(self, on: bool):
+  def _set_awake(self, on: bool, _ui_state=None):
     if on != self._awake:
-      DeviceSP._set_awake(on, ui_state)
+      super()._set_awake(on, _ui_state or ui_state)
       self._awake = on
       cloudlog.debug(f"setting display power {int(on)}")
       HARDWARE.set_display_power(on)

openpilot/sunnypilot/mads/tests/test_mads_state_machine.py

@@ -29,7 +29,7 @@ def make_event(event_types):
   event = {}
   for ev in event_types:
     event[ev] = NormalPermanentAlert("alert")
-  EVENTS_SP[0] = event
+  EVENTS_SP[0] = event  # type: ignore[assignment] # ty: ignore[invalid-assignment]
   return 0
 
 

openpilot/sunnypilot/mapd/mapd_manager.py

@@ -128,7 +128,7 @@ def main_thread():
     cloudlog.exception(f"mapd: failed to make {Paths.mapd_root()}")
 
   while True:
-    show_alert = get_files_for_cleanup() and params.get_bool("OsmLocal")
+    show_alert = bool(get_files_for_cleanup() and params.get_bool("OsmLocal"))
     set_offroad_alert("Offroad_OSMUpdateRequired", show_alert, "This alert will be cleared when new maps are downloaded.")
 
     update_osm_db()

openpilot/sunnypilot/modeld_v2/compile_modeld.py

@@ -129,7 +129,7 @@ def compile_split_policy(nv12: NV12Frame, model_w, model_h, prepare_only, frame_
 
   def random_inputs_run_fn(fn, seed, test_val=None, test_buffers=None, expect_match=True):
     input_queues, npy = make_split_input_queues(vision_input_shapes, policy_input_shapes, frame_skip, Device.DEFAULT)
-    np.random.seed(seed)
+    rng = np.random.default_rng(seed)
     Tensor.manual_seed(seed)
 
     testing = test_val is not None or test_buffers is not None
@@ -139,7 +139,7 @@ def compile_split_policy(nv12: NV12Frame, model_w, model_h, prepare_only, frame_
       frame = Tensor.randint(nv12.size, low=0, high=256, dtype='uint8').realize()
       big_frame = Tensor.randint(nv12.size, low=0, high=256, dtype='uint8').realize()
       for v in npy.values():
-        v[:] = np.random.randn(*v.shape).astype(v.dtype)
+        v[:] = rng.standard_normal(v.shape).astype(v.dtype)
       Device.default.synchronize()
       st = time.perf_counter()
       outs = fn(**input_queues, frame=frame, big_frame=big_frame)
@@ -319,11 +319,11 @@ def make_run_vision_multi_policy(vision_runner, policy_runners, nv12: NV12Frame,
 
 def _warmup_and_serialize(run_jit, input_queues, npy, nv12):
   for i in range(3):
-    np.random.seed(42 + i)
+    rng = np.random.default_rng(42 + i)
     frame = Tensor.randint(nv12.size, low=0, high=256, dtype='uint8').realize()
     big_frame = Tensor.randint(nv12.size, low=0, high=256, dtype='uint8').realize()
     for v in npy.values():
-      v[:] = np.random.randn(*v.shape).astype(v.dtype)
+      v[:] = rng.standard_normal(v.shape).astype(v.dtype)
     Device.default.synchronize()
     st = time.perf_counter()
     run_jit(**input_queues, frame=frame, big_frame=big_frame)

openpilot/sunnypilot/modeld_v2/parse_model_outputs.py

@@ -61,6 +61,7 @@ class Parser:
           weights[fidx] = weights[fidx][idxs]
           pred_mu[fidx] = pred_mu[fidx][idxs]
           pred_std[fidx] = pred_std[fidx][idxs]
+      assert out_shape is not None
       full_shape = tuple([raw.shape[0], in_N] + list(out_shape))
       outs[name + '_weights'] = weights
       outs[name + '_hypotheses'] = pred_mu.reshape(full_shape)
@@ -78,8 +79,10 @@ class Parser:
       pred_std_final = pred_std
 
     if out_N > 1:
+      assert out_shape is not None
       final_shape = tuple([raw.shape[0], out_N] + list(out_shape))
     else:
+      assert out_shape is not None
       final_shape = tuple([raw.shape[0],] + list(out_shape))
     outs[name] = pred_mu_final.reshape(final_shape)
     outs[name + '_stds'] = pred_std_final.reshape(final_shape)

openpilot/sunnypilot/modeld_v2/parse_model_outputs_split.py

@@ -65,6 +65,7 @@ class Parser:
           weights[fidx] = weights[fidx][idxs]
           pred_mu[fidx] = pred_mu[fidx][idxs]
           pred_std[fidx] = pred_std[fidx][idxs]
+      assert out_shape is not None
       full_shape = tuple([raw.shape[0], in_N] + list(out_shape))
       outs[name + '_weights'] = weights
       outs[name + '_hypotheses'] = pred_mu.reshape(full_shape)
@@ -82,8 +83,10 @@ class Parser:
       pred_std_final = pred_std
 
     if out_N > 1:
+      assert out_shape is not None
       final_shape = tuple([raw.shape[0], out_N] + list(out_shape))
     else:
+      assert out_shape is not None
       final_shape = tuple([raw.shape[0],] + list(out_shape))
     outs[name] = pred_mu_final.reshape(final_shape)
     outs[name + '_stds'] = pred_std_final.reshape(final_shape)

openpilot/sunnypilot/modeld_v2/tests/test_compile_modeld.py

@@ -61,7 +61,7 @@ class TestFrameSkipBufferLengthEquivalence:
 
 class TestTemporalSamplingEquivalence:
   def test_non20hz_desire_sampling_identity(self):
-    buf = np.random.randn(100, 1, 8).astype(np.float32)
+    buf = np.random.default_rng(0).standard_normal((100, 1, 8)).astype(np.float32)
     frame_skip = 1
     sampled = buf[::frame_skip].reshape(-1, 8)
     assert sampled.shape == (100, 8)
@@ -150,7 +150,7 @@ class TestOutputSlicePreservation:
   def test_vision_hidden_state_slice_used_for_features(self):
     mock_slices = {'hidden_state': slice(0, 512), 'plan': slice(512, 1024)}
     features_slice = mock_slices['hidden_state']
-    fake_output = np.random.randn(1, 1024).astype(np.float32)
+    fake_output = np.random.default_rng(0).standard_normal((1, 1024)).astype(np.float32)
     features = fake_output[:, features_slice]
     assert features.shape == (1, 512)
 

openpilot/sunnypilot/modeld_v2/tests/test_recovery_power.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 import numpy as np
 
 from openpilot.cereal import log
@@ -14,7 +16,7 @@ class MockStruct:
 
 
 def test_recovery_power_scaling():
-  state = MockStruct(
+  state: Any = MockStruct(
     PLANPLUS_CONTROL=0.75,
     LONG_SMOOTH_SECONDS=0.3,
     LAT_SMOOTH_SECONDS=0.1,
@@ -35,10 +37,10 @@ def test_recovery_power_scaling():
     recorded_curv_plans.append(plan.copy())
     return 0.0
 
-  modeld.get_accel_from_plan = mock_accel
-  modeld.get_curvature_from_output = mock_curvature
-  plan = np.random.rand(1, 100, 15).astype(np.float32)
-  planplus = np.random.rand(1, 100, 15).astype(np.float32)
+  modeld.get_accel_from_plan = mock_accel  # ty: ignore[invalid-assignment]
+  modeld.get_curvature_from_output = mock_curvature  # ty: ignore[invalid-assignment]
+  plan = np.random.default_rng(0).random((1, 100, 15)).astype(np.float32)
+  planplus = np.random.default_rng(1).random((1, 100, 15)).astype(np.float32)
   merged_plan = plan + planplus
 
   model_output: dict = {
@@ -60,7 +62,7 @@ def test_recovery_power_scaling():
     state.PLANPLUS_CONTROL = control
     recorded_vel.clear()
     recorded_curv_plans.clear()
-    ModelState.get_action_from_model(state, model_output, prev_action, 0.0, 0.0, v_ego)
+    ModelState.get_action_from_model(state, model_output, prev_action, 0.0, 0.0, v_ego)  # type: ignore[arg-type]
 
     expected_accel_plan_vel = plan[0, :, Plan.VELOCITY][:, 0] + planplus[0, :, Plan.VELOCITY][:, 0]
     np.testing.assert_allclose(recorded_vel[0], expected_accel_plan_vel, rtol=1e-5, atol=1e-6)

openpilot/sunnypilot/modeld_v2/warp.py

@@ -66,8 +66,8 @@ def compile_v2_warp(cam_w, cam_h, buffer_length):
 
   full_buffer = Tensor.zeros(img_buffer_shape, dtype='uint8').contiguous().realize()
   big_full_buffer = Tensor.zeros(img_buffer_shape, dtype='uint8').contiguous().realize()
-  new_frame_np = np.random.randint(0, 256, yuv_size, dtype=np.uint8)
-  new_big_frame_np = np.random.randint(0, 256, yuv_size, dtype=np.uint8)
+  new_frame_np = np.random.default_rng(0).integers(0, 256, yuv_size, dtype=np.uint8)
+  new_big_frame_np = np.random.default_rng(1).integers(0, 256, yuv_size, dtype=np.uint8)
   for i in range(10):
     img_inputs = [full_buffer,
                   Tensor.from_blob(new_frame_np.ctypes.data, (yuv_size,), dtype='uint8').realize(),
@@ -91,8 +91,8 @@ def compile_v2_warp(cam_w, cam_h, buffer_length):
   print(f"  Saved to {pkl_path}")
 
   jit = pickle.load(open(pkl_path, "rb"))
-  verify_frame = np.random.randint(0, 256, yuv_size, dtype=np.uint8)
-  verify_big_frame = np.random.randint(0, 256, yuv_size, dtype=np.uint8)
+  verify_frame = np.random.default_rng(0).integers(0, 256, yuv_size, dtype=np.uint8)
+  verify_big_frame = np.random.default_rng(1).integers(0, 256, yuv_size, dtype=np.uint8)
   fresh_inputs = [
     Tensor.zeros(img_buffer_shape, dtype='uint8').contiguous().realize(),
     Tensor.from_blob(verify_frame.ctypes.data, (yuv_size,), dtype='uint8').realize(),

openpilot/sunnypilot/models/helpers.py

@@ -126,7 +126,7 @@ def get_active_bundle(params: Params | None = None, raw_bundle_dict: dict | byte
   params = params or Params()
   try:
     active_bundle_dict = raw_bundle_dict if raw_bundle_dict is not None else (params.get("ModelManager_ActiveBundle") or {})
-    if active_bundle_dict and is_bundle_version_compatible(active_bundle_dict):
+    if isinstance(active_bundle_dict, dict) and active_bundle_dict and is_bundle_version_compatible(active_bundle_dict):
       return custom.ModelManagerSP.ModelBundle(**active_bundle_dict)
   except Exception:
     pass

openpilot/sunnypilot/models/manager.py

@@ -69,7 +69,7 @@ class ModelManagerSP:
         total_size = int(response.headers.get("content-length", 0))
         bytes_downloaded = 0
 
-        with open(path, 'wb') as f:
+        with open(path, 'wb') as f:  # noqa: ASYNC230
           async for chunk in response.content.iter_chunked(self._chunk_size):  # type: bytes
             f.write(chunk)
             bytes_downloaded += len(chunk)
@@ -110,7 +110,7 @@ class ModelManagerSP:
         async with session.get(chunk_url) as response:
           response.raise_for_status()
           chunk_size = int(response.headers.get("content-length", 0))
-          with open(chunk_path, 'wb') as f:
+          with open(chunk_path, 'wb') as f:  # noqa: ASYNC230
             async for data in response.content.iter_chunked(self._chunk_size):
               f.write(data)
               chunk_downloaded += len(data)
@@ -124,9 +124,9 @@ class ModelManagerSP:
               self._sync_artifact_progress(artifact)
               self._report_status()
 
-    with open(manifest_path, 'w') as f:
+    with open(manifest_path, 'w') as f:  # noqa: ASYNC230
       f.write(str(num_chunks))
-    if os.path.isfile(base_path):
+    if os.path.isfile(base_path):  # noqa: ASYNC240
       os.remove(base_path)
     del self._download_start_times[artifact.fileName]
 
@@ -165,7 +165,7 @@ class ModelManagerSP:
     except Exception as e:
       cloudlog.error(f"Error downloading {filename}: {str(e)}")
       for f in [full_path] + [p for p in (os.path.join(destination_path, f) for f in os.listdir(destination_path)) if filename in p]:
-        if os.path.isfile(f):
+        if os.path.isfile(f):  # noqa: ASYNC240
           os.remove(f)
       artifact.downloadProgress.status = custom.ModelManagerSP.DownloadStatus.failed
       artifact.downloadProgress.eta = 0
@@ -222,7 +222,8 @@ class ModelManagerSP:
       self.selected_bundle = None
 
     except Exception:
-      self.selected_bundle.status = custom.ModelManagerSP.DownloadStatus.failed
+      if self.selected_bundle is not None:
+        self.selected_bundle.status = custom.ModelManagerSP.DownloadStatus.failed
       raise
 
     finally:

openpilot/sunnypilot/selfdrive/controls/lib/dec/tests/pytest_dynamic_controller.py

@@ -84,7 +84,7 @@ def test_radarless_slowdown_triggers_blended(mock_cp, mock_mpc, default_sm):
   controller = DynamicExperimentalController(mock_cp, mock_mpc, params=MockParams())
 
   # Force conditions to simulate slowdown
-  controller._slow_down_filter = FakeKalman(value=1.0)  # Ensure urgency triggers slowdown
+  controller._slow_down_filter = FakeKalman(value=1.0)  # ty: ignore[invalid-assignment]
   controller._v_ego_kph = 35.0
   default_sm['modelV2'] = MockModelData(valid=False)  # Incomplete trajectory
 

openpilot/sunnypilot/selfdrive/controls/lib/nnlc/nnlc.py

@@ -78,7 +78,7 @@ class NeuralNetworkLateralControl(LatControlTorqueExtBase):
                                                                           self.torque_params, gravity_adjusted=False)
     torque_from_measurement = self.torque_from_lateral_accel_in_torque_space(LatControlInputs(self._measurement, self._roll_compensation, CS.vEgo, CS.aEgo),
                                                                              self.torque_params, gravity_adjusted=False)
-    self._pid_log.error = float(torque_from_setpoint - torque_from_measurement)
+    self._pid_log.error = float(torque_from_setpoint - torque_from_measurement)  # ty: ignore[invalid-assignment]
     self._ff = self.torque_from_lateral_accel_in_torque_space(LatControlInputs(self._gravity_adjusted_lateral_accel, self._roll_compensation,
                                                                                CS.vEgo, CS.aEgo), self.torque_params, gravity_adjusted=True)
     self._ff += get_friction_in_torque_space(self._desired_lateral_accel - self._actual_lateral_accel, self._lateral_accel_deadzone,
@@ -132,7 +132,7 @@ class NeuralNetworkLateralControl(LatControlTorqueExtBase):
                              + past_rolls + future_rolls
     torque_from_setpoint = self.model.evaluate(nnff_setpoint_input)
     torque_from_measurement = self.model.evaluate(nnff_measurement_input)
-    self._pid_log.error = torque_from_setpoint - torque_from_measurement
+    self._pid_log.error = torque_from_setpoint - torque_from_measurement  # ty: ignore[invalid-assignment]
 
     # The "pure" NNLC error response can be too weak for cars whose models were trained
     # with a lack of high-magnitude lateral acceleration data, for which the NNLC model
@@ -148,7 +148,7 @@ class NeuralNetworkLateralControl(LatControlTorqueExtBase):
       nnff_error_input = [CS.vEgo, self._setpoint - self._measurement, self.lateral_jerk_setpoint - self.lateral_jerk_measurement, 0.0]
       torque_from_error = self.model.evaluate(nnff_error_input)
       if sign(self._pid_log.error) == sign(torque_from_error) and abs(self._pid_log.error) < abs(torque_from_error):
-        self._pid_log.error = self._pid_log.error * (1.0 - error_blend_factor) + torque_from_error * error_blend_factor
+        self._pid_log.error = self._pid_log.error * (1.0 - error_blend_factor) + torque_from_error * error_blend_factor  # ty: ignore[invalid-assignment]
 
     # compute feedforward (same as nn setpoint output)
     friction_input = self.update_friction_input(self._setpoint, self._measurement)

openpilot/sunnypilot/selfdrive/controls/lib/smart_cruise_control/tests/test_vision_controller.py

@@ -4,6 +4,8 @@ Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
 This file is part of sunnypilot and is licensed under the MIT License.
 See the LICENSE.md file in the root directory for more details.
 """
+from typing import Any
+
 import numpy as np
 import pytest
 
@@ -113,7 +115,7 @@ class TestSmartCruiseControlVision:
     mdl = generate_modelV2()
     cs = generate_carState()
     controls_state = generate_controlsState()
-    self.sm = {'modelV2': mdl.modelV2, 'carState': cs.carState, 'controlsState': controls_state.controlsState}
+    self.sm: Any = {'modelV2': mdl.modelV2, 'carState': cs.carState, 'controlsState': controls_state.controlsState}
 
   def reset_params(self):
     self.params.put_bool("SmartCruiseControlVision", True, block=True)

openpilot/sunnypilot/selfdrive/controls/lib/speed_limit/speed_limit_resolver.py

@@ -152,9 +152,9 @@ class SpeedLimitResolver:
         self.distance_solutions[SpeedLimitSource.map] = distance_to_speed_limit_ahead
 
   def _get_source_solution_according_to_policy(self) -> custom.LongitudinalPlanSP.SpeedLimit.Source:
-    sources_for_policy = self._policy_to_sources_map[self.policy]
+    sources_for_policy = self._policy_to_sources_map[Policy(self.policy)]
 
-    if self.policy != Policy.combined:
+    if Policy(self.policy) != Policy.combined:
       # They are ordered in the order of preference, so we pick the first that's non-zero
       for source in sources_for_policy:
         if self.limit_solutions[source] > 0.:

openpilot/sunnypilot/selfdrive/locationd/tests/test_locationd.py

@@ -85,7 +85,7 @@ class TestLocationdProc:
     for msg in sorted(msgs, key=lambda x: x.logMonoTime):
       self.pm.send(msg.which(), msg)
       if msg.which() == "cameraOdometry":
-        self.pm.wait_for_readers_to_update(msg.which(), 0.1, dt=0.005)
+        self.pm.wait_for_readers_to_update(msg.which(), timeout=1, dt=0.005)
     time.sleep(1)  # wait for async params write
 
     lastGPS = json.loads(self.params.get('LastGPSPositionLLK'))

openpilot/sunnypilot/sunnylink/athena/tests/test_sunnylinkd.py

@@ -16,10 +16,10 @@ class TestSunnylinkdMethods:
     def mock_save_param(key, value, compression=False):
       self.saved_params.append((key, value, compression))
 
-    sunnylinkd.save_param_from_base64_encoded_string = mock_save_param
+    sunnylinkd.save_param_from_base64_encoded_string = mock_save_param  # ty: ignore[invalid-assignment]
 
   def teardown_method(self):
-    sunnylinkd.save_param_from_base64_encoded_string = self.original_save
+    sunnylinkd.save_param_from_base64_encoded_string = self.original_save  # ty: ignore[invalid-assignment]
 
   def test_saveParams_blocked(self):
     blocked_params = {

openpilot/sunnypilot/sunnylink/backups/utils.py

@@ -183,6 +183,6 @@ def transform_dict(obj):
 
 
 class SnakeCaseEncoder(json.JSONEncoder):
-  def encode(self, obj):
-    transformed_obj = transform_dict(obj)
+  def encode(self, o):
+    transformed_obj = transform_dict(o)
     return super().encode(transformed_obj)

openpilot/sunnypilot/sunnylink/tools/validate_settings_ui.py

@@ -129,6 +129,10 @@ def validate_rule(rule: dict, path: str, result: ValidationResult,
       return False
     valid = True
     for i, cond in enumerate(rule["conditions"]):
+      if not isinstance(cond, dict):
+        result.error("rule well-formedness", f"{path}.{rule_type}[{i}]: condition must be a dict")
+        valid = False
+        continue
       if not validate_rule(cond, f"{path}.{rule_type}[{i}]", result, capability_fields):
         valid = False
     return valid

openpilot/sunnypilot/sunnylink/utils.py

@@ -108,20 +108,22 @@ def _convert_param_to_type(value: bytes, param_type: ParamKeyType) -> bytes | st
   """
 
   # We convert to string anything that isn't bytes first. We later transform further.
-  if param_type != ParamKeyType.BYTES:
-    value = value.decode('utf-8')
+  if param_type == ParamKeyType.BYTES:
+    return value
+
+  decoded = value.decode('utf-8')
 
   if param_type == ParamKeyType.STRING:
-    value = value
+    return decoded
   elif param_type == ParamKeyType.BOOL:
-    value = value.lower() in ('true', '1', 'yes')
+    return decoded.lower() in ('true', '1', 'yes')
   elif param_type == ParamKeyType.INT:
-    value = int(value)
+    return int(decoded)
   elif param_type == ParamKeyType.FLOAT:
-    value = float(value)
+    return float(decoded)
   elif param_type == ParamKeyType.TIME:
-    value = str(value)
+    return str(decoded)
   elif param_type == ParamKeyType.JSON:
-    value = json.loads(value)
+    return json.loads(decoded)
 
-  return value
+  return decoded

openpilot/sunnypilot/system/sensord/tests/test_sensord.py

@@ -71,7 +71,7 @@ ALL_SENSORS = {
 }
 
 
-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)
@@ -181,7 +181,7 @@ class TestSensord:
   def test_logmonottime_timestamp_diff(self):
     # ensure diff between the message logMonotime and sample timestamp is small
 
-    tdiffs = list()
+    tdiffs = []
     for etype in self.events:
       for measurement in self.events[etype]:
         m = getattr(measurement, measurement.which())
@@ -203,7 +203,7 @@ class TestSensord:
     assert avg_diff < 4, f"Avg packet diff: {avg_diff:.1f}ms"
 
   def test_sensor_values(self):
-    sensor_values = dict()
+    sensor_values = {}
     for etype in self.events:
       for measurement in self.events[etype]:
         m = getattr(measurement, measurement.which())

openpilot/system/athena/athenad.py

@@ -679,18 +679,17 @@ def add_log_to_queue(log_path, log_id, is_sunnylink=False):
                      f"after compression: {compressed_size} bytes, " +
                      f"after encoding: {encoded_size} bytes")
 
-    jsonrpc = {
+    params: dict[str, str | bool] = {"logs": payload}
+    if is_sunnylink and is_compressed:
+      params["compressed"] = is_compressed
+
+    jsonrpc: dict = {
       "method": "forwardLogs",
-      "params": {
-        "logs": payload
-      },
+      "params": params,
       "jsonrpc": "2.0",
       "id": log_id
     }
 
-    if is_sunnylink and is_compressed:
-      jsonrpc["params"]["compressed"] = is_compressed
-
     jsonrpc_str = json.dumps(jsonrpc)
     size_in_bytes = len(jsonrpc_str.encode('utf-8'))
 
@@ -783,18 +782,17 @@ def stat_handler(end_event: threading.Event, stats_dir=None, is_sunnylink=False)
               payload = base64.b64encode(compressed_data).decode()
               is_compressed = True
 
-            jsonrpc = {
+            params: dict[str, str | bool] = {"stats": payload}
+            if is_sunnylink and is_compressed:
+              params["compressed"] = is_compressed
+
+            jsonrpc: dict = {
               "method": "storeStats",
-              "params": {
-                "stats": payload
-              },
+              "params": params,
               "jsonrpc": "2.0",
               "id": stat_filenames[0]
             }
 
-            if is_sunnylink and is_compressed:
-              jsonrpc["params"]["compressed"] = is_compressed
-
             send_queue_push(json.dumps(jsonrpc), SEND_PRIORITY_LOW)
           os.remove(stat_path)
         last_scan = curr_scan

openpilot/system/ui/sunnypilot/widgets/list_view.py

@@ -4,7 +4,7 @@ Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
 This file is part of sunnypilot and is licensed under the MIT License.
 See the LICENSE.md file in the root directory for more details.
 """
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 
 import pyray as rl
 from openpilot.common.params import Params
@@ -126,7 +126,7 @@ class DualButtonActionSP(DualButtonAction):
 
 
 class MultipleButtonActionSP(MultipleButtonAction):
-  def __init__(self, buttons: list[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable | None = None,
+  def __init__(self, buttons: Sequence[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable | None = None,
                param: str | None = None):
     MultipleButtonAction.__init__(self, buttons, button_width, selected_index, callback)
     self.param_key = param
@@ -366,7 +366,7 @@ def toggle_item_sp(title: str | Callable[[], str], description: str | Callable[[
   return ListItemSP(title=title, description=description, action_item=action, icon=icon)
 
 
-def multiple_button_item_sp(title: str | Callable[[], str], description: str | Callable[[], str], buttons: list[str | Callable[[], str]],
+def multiple_button_item_sp(title: str | Callable[[], str], description: str | Callable[[], str], buttons: Sequence[str | Callable[[], str]],
                             selected_index: int = 0, button_width: int = style.BUTTON_ACTION_WIDTH, callback: Callable | None = None,
                             icon: str = "", param: str | None = None, inline: bool = False) -> ListItemSP:
   action = MultipleButtonActionSP(buttons, button_width, selected_index, callback=callback, param=param)

openpilot/system/ui/sunnypilot/widgets/tree_dialog.py

@@ -48,9 +48,9 @@ class TreeItemWidget(Button):
     self.border_radius = 10
     self.is_expanded = is_expanded
 
-  def _render(self, rect):
+  def _render(self, _):
     indent = 60 * self.indent_level
-    self._rect = rl.Rectangle(rect.x + indent, rect.y, rect.width - indent, rect.height)
+    self._rect = rl.Rectangle(_.x + indent, _.y, _.width - indent, _.height)
     if self.is_pressed:
       color = BUTTON_PRESSED_BACKGROUND_COLORS[self._button_style]
     elif self.selected and self.ref != "search_bar":

openpilot/system/ui/widgets/list_view.py

@@ -1,6 +1,6 @@
 import os
 import pyray as rl
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from abc import ABC
 from openpilot.system.ui.lib.application import gui_app, FontWeight, MousePos
 from openpilot.system.ui.lib.multilang import tr
@@ -207,7 +207,7 @@ class DualButtonAction(ItemAction):
 
 
 class MultipleButtonAction(ItemAction):
-  def __init__(self, buttons: list[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable | None = None):
+  def __init__(self, buttons: Sequence[str | Callable[[], str]], button_width: int, selected_index: int = 0, callback: Callable | None = None):
     super().__init__(width=len(buttons) * button_width + (len(buttons) - 1) * RIGHT_ITEM_PADDING, enabled=True)
     self.buttons = buttons
     self.button_width = button_width

openpilot/system/ui/widgets/network.py

@@ -1,6 +1,6 @@
 from enum import IntEnum
 from functools import partial
-from typing import cast
+from typing import Any, cast
 
 import pyray as rl
 from openpilot.system.ui.lib.application import gui_app
@@ -27,9 +27,9 @@ try:
   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
+  Params: Any = None
+  ui_state: Any = None
+  PrimeType: Any = None
 
 NM_DEVICE_STATE_NEED_AUTH = 60
 MIN_PASSWORD_LENGTH = 8