sm

selfdrive/controls/tests/test_conditional_chill_mode.py

@@ -40,6 +40,14 @@ class FakeDetector:
     return None
 
 
+class FakeSubMaster:
+  def __init__(self, services):
+    self.services = dict(services)
+
+  def __getitem__(self, key):
+    return self.services[key]
+
+
 def make_sm():
   return {
     "carState": SimpleNamespace(standstill=False, leftBlinker=False, rightBlinker=False),
@@ -51,6 +59,10 @@ def make_sm():
   }
 
 
+def make_submaster_sm():
+  return FakeSubMaster(make_sm())
+
+
 def make_toggles():
   return SimpleNamespace(
     conditional_chill_speed=45 * CV.MPH_TO_MS,
@@ -224,3 +236,16 @@ def test_ccm_restores_persisted_manual_experimental_override(monkeypatch):
   assert ccm.experimental_mode
   assert ccm.status_value == CCStatus["USER_EXPERIMENTAL"]
   assert planner.params_memory.get_int("CCStatus") == CCStatus["USER_EXPERIMENTAL"]
+
+
+def test_ccm_adjacent_lead_veto_works_with_submaster_like_input(monkeypatch):
+  planner, _detector, ccm = make_ccm()
+  sm = make_submaster_sm()
+  toggles = make_toggles()
+  sm["starpilotRadarState"].leadLeft = SimpleNamespace(status=True, dRel=20.0, vLead=12.0)
+
+  monkeypatch.setattr("openpilot.starpilot.controls.lib.conditional_chill_mode.time.monotonic", lambda: 10.0)
+  ccm.update(60 * CV.MPH_TO_MS, 65 * CV.MPH_TO_MS, sm, toggles)
+
+  assert ccm.experimental_mode
+  assert ccm.status_value == CCStatus["OFF"]

starpilot/controls/lib/conditional_chill_mode.py

@@ -194,7 +194,7 @@ class ConditionalChillMode:
     return CCStatus["LEAD"]
 
   def _adjacent_lead_ambiguous(self, sm, v_ego):
-    radar_state = sm.get("starpilotRadarState")
+    radar_state = self._get_sm_service(sm, "starpilotRadarState")
     if radar_state is None:
       return False
 
@@ -208,6 +208,16 @@ class ConditionalChillMode:
 
     return False
 
+  @staticmethod
+  def _get_sm_service(sm, key):
+    if isinstance(sm, dict):
+      return sm.get(key)
+
+    try:
+      return sm[key]
+    except (KeyError, IndexError, TypeError, AttributeError):
+      return None
+
   def _write_status(self, status_value):
     if status_value != self._prev_cc_status:
       self.params_memory.put_int("CCStatus", status_value)