selfdrive/car: ban cereal and capnp

.importlinter

@@ -1,13 +1,17 @@
 [importlinter]
 root_packages =
     openpilot
+    cereal
+    capnp
 
 [importlinter:contract:1]
 name = Forbid imports from openpilot.selfdrive.car to openpilot.system
 type = forbidden
 source_modules =
     openpilot.selfdrive.car
 forbidden_modules =
+    cereal
+    capnp
     openpilot.common
     openpilot.selfdrive.controls
     openpilot.selfdrive.debug
@@ -44,5 +48,16 @@ ignore_imports =
     openpilot.selfdrive.car.tests.test_car_interfaces -> openpilot.selfdrive.pandad
     openpilot.selfdrive.car.tests.test_models -> openpilot.selfdrive.pandad
     openpilot.selfdrive.car.tests.test_car_interfaces -> openpilot.selfdrive.test.fuzzy_generation
+    openpilot.selfdrive.car.tests.test_models -> capnp
+    openpilot.selfdrive.car.tests.test_car_interfaces -> cereal
+    openpilot.selfdrive.car.tests.test_car_interfaces -> cereal.messaging
+    openpilot.selfdrive.car.tests.test_car_interfaces -> openpilot.selfdrive.test.fuzzy_generation
+    openpilot.selfdrive.car.tests.test_models -> cereal
+    openpilot.selfdrive.car.tests.test_models -> cereal.messaging
+    openpilot.selfdrive.car.card -> cereal
+    openpilot.selfdrive.car.card -> cereal.messaging
     openpilot.selfdrive.car.car_specific -> openpilot.selfdrive.controls.lib.events
+    openpilot.selfdrive.car.car_specific -> cereal
+    openpilot.selfdrive.car.car_specific -> cereal.messaging
+    openpilot.selfdrive.car.card -> capnp
 unmatched_ignore_imports_alerting = warn

selfdrive/car/__init__.py

@@ -5,10 +5,8 @@
 from enum import IntFlag, ReprEnum, EnumType
 from dataclasses import replace
 
-import capnp
-
-from cereal import car
 from panda.python.uds import SERVICE_TYPE
+from openpilot.selfdrive.car import structs
 from openpilot.selfdrive.car.can_definitions import CanData
 from openpilot.selfdrive.car.docs_definitions import CarDocs
 from openpilot.selfdrive.car.common.numpy_fast import clip, interp
@@ -23,7 +21,7 @@
 # kg of standard extra cargo to count for drive, gas, etc...
 STD_CARGO_KG = 136.
 
-ButtonType = car.CarState.ButtonEvent.Type
+ButtonType = structs.CarState.ButtonEvent.Type
 AngleRateLimit = namedtuple('AngleRateLimit', ['speed_bp', 'angle_v'])
 
 
@@ -35,18 +33,18 @@ 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, structs.CarState.ButtonEvent.Type],
+                         unpressed_btn: int = 0) -> list[structs.CarState.ButtonEvent]:
+  events: list[structs.CarState.ButtonEvent] = []
 
   if cur_btn == prev_btn:
     return events
 
   # Add events for button presses, multiple when a button switches without going to unpressed
   for pressed, btn in ((False, prev_btn), (True, cur_btn)):
     if btn != unpressed_btn:
-      events.append(car.CarState.ButtonEvent(pressed=pressed,
-                                             type=buttons_dict.get(btn, ButtonType.unknown)))
+      events.append(structs.CarState.ButtonEvent(pressed=pressed,
+                                                 type=buttons_dict.get(btn, ButtonType.unknown)))
   return events
 
 
@@ -183,7 +181,7 @@ def rate_limit(new_value, last_value, dw_step, up_step):
 
 
 def get_friction(lateral_accel_error: float, lateral_accel_deadzone: float, friction_threshold: float,
-                 torque_params: car.CarParams.LateralTorqueTuning, friction_compensation: bool) -> float:
+                 torque_params: structs.CarParams.LateralTorqueTuning, friction_compensation: bool) -> float:
   friction_interp = interp(
     apply_center_deadzone(lateral_accel_error, lateral_accel_deadzone),
     [-friction_threshold, friction_threshold],
@@ -203,8 +201,8 @@ def make_tester_present_msg(addr, bus, subaddr=None, suppress_response=False):
   return CanData(addr, bytes(dat), bus)
 
 
-def get_safety_config(safety_model, safety_param = None):
-  ret = car.CarParams.SafetyConfig.new_message()
+def get_safety_config(safety_model: structs.CarParams.SafetyModel, safety_param: int = None) -> structs.CarParams.SafetyConfig:
+  ret = structs.CarParams.SafetyConfig()
   ret.safetyModel = safety_model
   if safety_param is not None:
     ret.safetyParam = safety_param

selfdrive/car/body/carcontroller.py

@@ -1,3 +1,4 @@
+import copy
 import numpy as np
 from numbers import Number
 
@@ -132,7 +133,7 @@ def update(self, CC, CS, now_nanos):
     can_sends = []
     can_sends.append(bodycan.create_control(self.packer, torque_l, torque_r))
 
-    new_actuators = CC.actuators.as_builder()
+    new_actuators = copy.copy(CC.actuators)
     new_actuators.accel = torque_l
     new_actuators.steer = torque_r
     new_actuators.steerOutputCan = torque_r

selfdrive/car/body/carstate.py

@@ -1,12 +1,12 @@
-from cereal import car
 from opendbc.can.parser import CANParser
+from openpilot.selfdrive.car import structs
 from openpilot.selfdrive.car.interfaces import CarStateBase
 from openpilot.selfdrive.car.body.values import DBC
 
 
 class CarState(CarStateBase):
-  def update(self, cp, *_):
-    ret = car.CarState.new_message()
+  def update(self, cp, *_) -> structs.CarState:
+    ret = structs.CarState()
 
     ret.wheelSpeeds.fl = cp.vl['MOTORS_DATA']['SPEED_L']
     ret.wheelSpeeds.fr = cp.vl['MOTORS_DATA']['SPEED_R']
@@ -23,7 +23,7 @@ def update(self, cp, *_):
     ret.fuelGauge = cp.vl["BODY_DATA"]["BATT_PERCENTAGE"] / 100
 
     # irrelevant for non-car
-    ret.gearShifter = car.CarState.GearShifter.drive
+    ret.gearShifter = structs.CarState.GearShifter.drive
     ret.cruiseState.enabled = True
     ret.cruiseState.available = True
 

selfdrive/car/body/fingerprints.py

@@ -1,8 +1,8 @@
 # ruff: noqa: E501
-from cereal import car
+from openpilot.selfdrive.car.structs import CarParams
 from openpilot.selfdrive.car.body.values import CAR
 
-Ecu = car.CarParams.Ecu
+Ecu = CarParams.Ecu
 
 # debug ecu fw version is the git hash of the firmware
 

selfdrive/car/body/interface.py

@@ -1,16 +1,15 @@
 import math
-from cereal import car
-from openpilot.selfdrive.car import get_safety_config
+from openpilot.selfdrive.car import get_safety_config, structs
 from openpilot.selfdrive.car.interfaces import CarInterfaceBase
 from openpilot.selfdrive.car.body.values import SPEED_FROM_RPM
 
 
 class CarInterface(CarInterfaceBase):
   @staticmethod
-  def _get_params(ret, candidate, fingerprint, car_fw, experimental_long, docs):
+  def _get_params(ret: structs.CarParams, candidate, fingerprint, car_fw, experimental_long, docs) -> structs.CarParams:
     ret.notCar = True
     ret.carName = "body"
-    ret.safetyConfigs = [get_safety_config(car.CarParams.SafetyModel.body)]
+    ret.safetyConfigs = [get_safety_config(structs.CarParams.SafetyModel.body)]
 
     ret.minSteerSpeed = -math.inf
     ret.maxLateralAccel = math.inf  # TODO: set to a reasonable value
@@ -21,6 +20,6 @@ def _get_params(ret, candidate, fingerprint, car_fw, experimental_long, docs):
 
     ret.radarUnavailable = True
     ret.openpilotLongitudinalControl = True
-    ret.steerControlType = car.CarParams.SteerControlType.angle
+    ret.steerControlType = structs.CarParams.SteerControlType.angle
 
     return ret

selfdrive/car/body/values.py

@@ -1,9 +1,9 @@
-from cereal import car
 from openpilot.selfdrive.car import CarSpecs, PlatformConfig, Platforms, dbc_dict
+from openpilot.selfdrive.car.structs import CarParams
 from openpilot.selfdrive.car.docs_definitions import CarDocs
 from openpilot.selfdrive.car.fw_query_definitions import FwQueryConfig, Request, StdQueries
 
-Ecu = car.CarParams.Ecu
+Ecu = CarParams.Ecu
 
 SPEED_FROM_RPM = 0.008587
 

selfdrive/car/car_helpers.py

@@ -1,9 +1,9 @@
 import os
 import time
 
-from cereal import car
 from openpilot.selfdrive.car import carlog, gen_empty_fingerprint
 from openpilot.selfdrive.car.can_definitions import CanRecvCallable, CanSendCallable
+from openpilot.selfdrive.car.structs import CarParams
 from openpilot.selfdrive.car.fingerprints import eliminate_incompatible_cars, all_legacy_fingerprint_cars
 from openpilot.selfdrive.car.fw_versions import ObdCallback, get_fw_versions_ordered, get_present_ecus, match_fw_to_car
 from openpilot.selfdrive.car.interfaces import get_interface_attr
@@ -82,7 +82,7 @@ def can_fingerprint(can_recv: CanRecvCallable) -> tuple[str | None, dict[int, di
 
 # **** for use live only ****
 def fingerprint(can_recv: CanRecvCallable, can_send: CanSendCallable, set_obd_multiplexing: ObdCallback, num_pandas: int,
-                cached_params: type[car.CarParams] | None) -> tuple[str | None, dict, str, list, int, bool]:
+                cached_params: CarParams | None) -> tuple[str | None, dict, str, list[CarParams.CarFw], CarParams.FingerprintSource, bool]:
   fixed_fingerprint = os.environ.get('FINGERPRINT', "")
   skip_fw_query = os.environ.get('SKIP_FW_QUERY', False)
   disable_fw_cache = os.environ.get('DISABLE_FW_CACHE', False)
@@ -129,17 +129,17 @@ def fingerprint(can_recv: CanRecvCallable, can_send: CanSendCallable, set_obd_mu
   car_fingerprint, finger = can_fingerprint(can_recv)
 
   exact_match = True
-  source = car.CarParams.FingerprintSource.can
+  source = CarParams.FingerprintSource.can
 
   # If FW query returns exactly 1 candidate, use it
   if len(fw_candidates) == 1:
     car_fingerprint = list(fw_candidates)[0]
-    source = car.CarParams.FingerprintSource.fw
+    source = CarParams.FingerprintSource.fw
     exact_match = exact_fw_match
 
   if fixed_fingerprint:
     car_fingerprint = fixed_fingerprint
-    source = car.CarParams.FingerprintSource.fixed
+    source = CarParams.FingerprintSource.fixed
 
   carlog.error({"event": "fingerprinted", "car_fingerprint": str(car_fingerprint), "source": source, "fuzzy": not exact_match,
                 "cached": cached, "fw_count": len(car_fw), "ecu_responses": list(ecu_rx_addrs), "vin_rx_addr": vin_rx_addr,
@@ -148,21 +148,21 @@ def fingerprint(can_recv: CanRecvCallable, can_send: CanSendCallable, set_obd_mu
   return car_fingerprint, finger, vin, car_fw, source, exact_match
 
 
-def get_car_interface(CP):
+def get_car_interface(CP: CarParams):
   CarInterface, CarController, CarState = interfaces[CP.carFingerprint]
   return CarInterface(CP, CarController, CarState)
 
 
 def get_car(can_recv: CanRecvCallable, can_send: CanSendCallable, set_obd_multiplexing: ObdCallback, experimental_long_allowed: bool,
-            num_pandas: int = 1, cached_params: type[car.CarParams] | None = None):
+            num_pandas: int = 1, cached_params: CarParams | None = None):
   candidate, fingerprints, vin, car_fw, source, exact_match = fingerprint(can_recv, can_send, set_obd_multiplexing, num_pandas, cached_params)
 
   if candidate is None:
     carlog.error({"event": "car doesn't match any fingerprints", "fingerprints": repr(fingerprints)})
     candidate = "MOCK"
 
   CarInterface, _, _ = interfaces[candidate]
-  CP = CarInterface.get_params(candidate, fingerprints, car_fw, experimental_long_allowed, docs=False)
+  CP: CarParams = CarInterface.get_params(candidate, fingerprints, car_fw, experimental_long_allowed, docs=False)
   CP.carVin = vin
   CP.carFw = car_fw
   CP.fingerprintSource = source

selfdrive/car/car_specific.py

@@ -1,16 +1,16 @@
 from cereal import car
 import cereal.messaging as messaging
-from openpilot.selfdrive.car import DT_CTRL
-from openpilot.selfdrive.car.interfaces import MAX_CTRL_SPEED
+from openpilot.selfdrive.car import DT_CTRL, structs
+from openpilot.selfdrive.car.interfaces import MAX_CTRL_SPEED, CarStateBase, CarControllerBase
 from openpilot.selfdrive.car.volkswagen.values import CarControllerParams as VWCarControllerParams
 from openpilot.selfdrive.car.hyundai.interface import ENABLE_BUTTONS as HYUNDAI_ENABLE_BUTTONS
 
 from openpilot.selfdrive.controls.lib.events import Events
 
-ButtonType = car.CarState.ButtonEvent.Type
-GearShifter = car.CarState.GearShifter
+ButtonType = structs.CarState.ButtonEvent.Type
+GearShifter = structs.CarState.GearShifter
 EventName = car.CarEvent.EventName
-NetworkLocation = car.CarParams.NetworkLocation
+NetworkLocation = structs.CarParams.NetworkLocation
 
 
 # TODO: the goal is to abstract this file into the CarState struct and make events generic
@@ -29,15 +29,15 @@ def update(self, CS: car.CarState):
 
 
 class CarSpecificEvents:
-  def __init__(self, CP: car.CarParams):
+  def __init__(self, CP: structs.CarParams):
     self.CP = CP
 
     self.steering_unpressed = 0
     self.low_speed_alert = False
     self.no_steer_warning = False
     self.silent_steer_warning = True
 
-  def update(self, CS, CS_prev, CC, CC_prev):
+  def update(self, CS: CarStateBase, CS_prev: car.CarState, CC: CarControllerBase, CC_prev: car.CarControl):
     if self.CP.carName in ('body', 'mock'):
       events = Events()
 
@@ -50,15 +50,15 @@ def update(self, CS, CS_prev, CC, CC_prev):
     elif self.CP.carName == 'nissan':
       events = self.create_common_events(CS.out, CS_prev, extra_gears=[GearShifter.brake])
 
-      if CS.lkas_enabled:
+      if CS.lkas_enabled:  # type: ignore[attr-defined]
         events.add(EventName.invalidLkasSetting)
 
     elif self.CP.carName == 'mazda':
       events = self.create_common_events(CS.out, CS_prev)
 
-      if CS.lkas_disabled:
+      if CS.lkas_disabled:  # type: ignore[attr-defined]
         events.add(EventName.lkasDisabled)
-      elif CS.low_speed_alert:
+      elif CS.low_speed_alert:  # type: ignore[attr-defined]
         events.add(EventName.belowSteerSpeed)
 
     elif self.CP.carName == 'chrysler':
@@ -99,7 +99,7 @@ def update(self, CS, CS_prev, CC, CC_prev):
       if self.CP.openpilotLongitudinalControl:
         if CS.out.cruiseState.standstill and not CS.out.brakePressed:
           events.add(EventName.resumeRequired)
-        if CS.low_speed_lockout:
+        if CS.low_speed_lockout:  # type: ignore[attr-defined]
           events.add(EventName.lowSpeedLockout)
         if CS.out.vEgo < self.CP.minEnableSpeed:
           events.add(EventName.belowEngageSpeed)
@@ -121,7 +121,7 @@ def update(self, CS, CS_prev, CC, CC_prev):
 
       # Enabling at a standstill with brake is allowed
       # TODO: verify 17 Volt can enable for the first time at a stop and allow for all GMs
-      below_min_enable_speed = CS.out.vEgo < self.CP.minEnableSpeed or CS.moving_backward
+      below_min_enable_speed = CS.out.vEgo < self.CP.minEnableSpeed or CS.moving_backward  # type: ignore[attr-defined]
       if below_min_enable_speed and not (CS.out.standstill and CS.out.brake >= 20 and
                                          self.CP.networkLocation == NetworkLocation.fwdCamera):
         events.add(EventName.belowEngageSpeed)
@@ -149,14 +149,14 @@ def update(self, CS, CS_prev, CC, CC_prev):
         if CC_prev.enabled and CS.out.vEgo < self.CP.minEnableSpeed:
           events.add(EventName.speedTooLow)
 
-      if CC.eps_timer_soft_disable_alert:
+      if CC.eps_timer_soft_disable_alert:  # type: ignore[attr-defined]
         events.add(EventName.steerTimeLimit)
 
     elif self.CP.carName == 'hyundai':
       # On some newer model years, the CANCEL button acts as a pause/resume button based on the PCM state
       # To avoid re-engaging when openpilot cancels, check user engagement intention via buttons
       # Main button also can trigger an engagement on these cars
-      allow_enable = any(btn in HYUNDAI_ENABLE_BUTTONS for btn in CS.cruise_buttons) or any(CS.main_buttons)
+      allow_enable = any(btn in HYUNDAI_ENABLE_BUTTONS for btn in CS.cruise_buttons) or any(CS.main_buttons)  # type: ignore[attr-defined]
       events = self.create_common_events(CS.out, CS_prev, pcm_enable=self.CP.pcmCruise, allow_enable=allow_enable)
 
       # low speed steer alert hysteresis logic (only for cars with steer cut off above 10 m/s)
@@ -172,8 +172,8 @@ def update(self, CS, CS_prev, CC, CC_prev):
 
     return events
 
-  def create_common_events(self, CS, CS_prev, extra_gears=None, pcm_enable=True, allow_enable=True,
-                           enable_buttons=(ButtonType.accelCruise, ButtonType.decelCruise)):
+  def create_common_events(self, CS: structs.CarState, CS_prev: car.CarState, extra_gears=None, pcm_enable=True,
+                           allow_enable=True, enable_buttons=(ButtonType.accelCruise, ButtonType.decelCruise)):
     events = Events()
 
     if CS.doorOpen: