Keep HKG angle AOL armed through temporary stop restrictions

openpilot/starpilot/aol/tests/test_angle_stop_fault_guards.py

@@ -0,0 +1,51 @@
+from contextlib import ExitStack
+
+import pytest
+
+from opendbc.car.hyundai.values import CAR
+from openpilot.common.params import Params
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.aol.intent import read_settings
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+
+
+@pytest.mark.parametrize('identity', [CAR.HYUNDAI_IONIQ_5_PE, CAR.KIA_EV9])
+@pytest.mark.parametrize('fault', ['rolling_denial', 'invalid_native', 'invalid_can', 'permanent_eps', 'panda_rx_fault'])
+def test_temporary_stop_intent_retention_cannot_mask_real_faults(identity, fault):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    with OpenpilotPrefix():
+      saved = Params()
+      saved.put_bool('AlwaysOnLateral', True, block=True)
+      saved.put('LKASButtonControl', 9, block=True)
+      saved.put('MainCruiseButtonControl', 0, block=True)
+      saved.put('AolBrakePauseSpeedMps', 0.0, block=True)
+      for key in ('CancelButtonControl', 'LongCancelButtonControl', 'VeryLongCancelButtonControl'):
+        saved.put(key, 0, block=True)
+      sd, _, step, card, receive = case._ioniq_availability_callers(full_op=False, stock_angle=identity)
+      sd.aol_settings = read_settings(saved)
+      assert sd.aol_settings == card.aol_card_intent.settings
+      rolling = fixture.car_state()
+      rolling.vEgo = rolling.vEgoRaw = 1.0
+      axis, _, command = step(2, observed_state=rolling, panda_controls_allowed=False)
+      assert axis.lateralActive and command.latActive
+      assert receive(10_020_000_000, state=rolling, axis=sd.pm.messages['aolAxisState'],
+                     event=sd.pm.messages['onroadEvents'])
+      state = fixture.car_state()
+      state.vEgo = state.vEgoRaw = 1.0 if fault == 'rolling_denial' else 0.0
+      state.standstill = fault != 'rolling_denial'
+      state.canValid = fault != 'invalid_can'
+      state.steerFaultPermanent = fault == 'permanent_eps'
+      axis, _, command = step(3, observed_state=state, allowed=False, native_request_lat=True,
+                              native_request_long=False, panda_controls_allowed=False,
+                              native_valid=fault != 'invalid_native', panda_fault=fault == 'panda_rx_fault')
+      assert not axis.lateralActive and not command.latActive
+      assert not axis.longitudinalActive and not command.longActive
+      assert not receive(10_030_000_000, state=state, axis=sd.pm.messages['aolAxisState'],
+                         event=sd.pm.messages['onroadEvents'])
+      assert not card.aol_card_intent.allowed_latch
+      axis, _, command = step(4, observed_state=rolling, panda_controls_allowed=False)
+      assert not axis.lateralActive and not command.latActive
+      assert not axis.longitudinalActive and not command.longActive
+      assert not card.aol_card_intent.allowed_latch

openpilot/starpilot/aol/tests/test_angle_stop_native_feedback.py

@@ -0,0 +1,65 @@
+from contextlib import ExitStack
+
+import pytest
+
+from opendbc.car.hyundai.canfd_angle_aol import temporary_restriction, MODELS
+from openpilot.common.params import Params
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.aol.intent import read_settings
+from openpilot.starpilot.aol.vehicle import native_latch_rejected
+from openpilot.starpilot.aol.runtime import current_native
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+
+
+@pytest.mark.parametrize('identity', sorted(MODELS))
+@pytest.mark.parametrize('signal', [False, True])
+@pytest.mark.parametrize('brake', [False, True])
+def test_native_stop_restriction_withdraws_steering_without_disarming_driver_intent(identity, signal, brake):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    with OpenpilotPrefix():
+      saved = Params()
+      saved.put_bool('AlwaysOnLateral', True, block=True)
+      saved.put('LKASButtonControl', 9, block=True)
+      saved.put('MainCruiseButtonControl', 0, block=True)
+      saved.put('AolBrakePauseSpeedMps', 0.0, block=True)
+      saved.put('PauseLateralSpeed', 0.0, block=True)
+      saved.put_bool('PauseLateralOnSignal', False, block=True)
+      saved.put('LateralResumeDelay', 0.0, block=True)
+      for key in ('CancelButtonControl', 'LongCancelButtonControl', 'VeryLongCancelButtonControl'):
+        saved.put(key, 0, block=True)
+      sd, _, step, card, receive = case._ioniq_availability_callers(full_op=False, stock_angle=identity)
+      sd.aol_settings = read_settings(saved)
+      assert sd.aol_settings == card.aol_card_intent.settings
+      assert sd.aol_settings.pause_brake_mps == 0.0
+      rolling = fixture.car_state()
+      rolling.vEgo = rolling.vEgoRaw = 1.0
+      axis, _, command = step(2, observed_state=rolling, panda_controls_allowed=False)
+      assert axis.lateralActive and command.latActive
+      assert receive(10_020_000_000, state=rolling, axis=sd.pm.messages['aolAxisState'],
+                     event=sd.pm.messages['onroadEvents'])
+      for tick in range(3, 37):
+        stopped = fixture.car_state(brake=brake)
+        stopped.vEgo = stopped.vEgoRaw = 0.0
+        stopped.standstill = True
+        stopped.leftBlinker = signal
+        assert temporary_restriction(card.CP, stopped)
+        axis, _, command = step(tick, observed_state=stopped, allowed=False, native_request_lat=True,
+                                native_request_long=False, panda_controls_allowed=False)
+        now = 10_000_000_000 + tick * 10_000_000
+        native = current_native(sd.sm, sd.CP, now_ns=now, axis_session_id=sd.aol_session_id)
+        assert native is not None and native.requestedLateral and not native.lateralAllowed
+        assert not native_latch_rejected(card.CP, native, state=stopped)
+        assert not axis.lateralActive and not command.latActive
+        assert not axis.longitudinalActive and not command.longActive
+        assert receive(now, state=stopped, axis=sd.pm.messages['aolAxisState'],
+                       event=sd.pm.messages['onroadEvents']), card.aol_card_intent.latch_clear_reason
+        assert card.aol_card_intent.allowed_latch and not sd.aol_authority_lost
+      axis, _, command = step(37, observed_state=rolling, allowed=True, native_request_lat=True,
+                              native_request_long=False, panda_controls_allowed=False)
+      assert axis.lateralActive and command.latActive
+      assert not axis.longitudinalActive and not command.longActive
+      assert receive(10_370_000_000, state=rolling, axis=sd.pm.messages['aolAxisState'],
+                     event=sd.pm.messages['onroadEvents'])
+      assert card.aol_card_intent.allowed_latch

openpilot/starpilot/aol/transport_pause.py

@@ -12,13 +12,13 @@ def cp_identity(cp):
           tuple((int(c.safetyModel.raw), int(c.safetyParam)) for c in cp.safetyConfigs))
 
 
-def _configured_transport_native(sm, cp, session, now_ns, native=None, *, require_longitudinal=True):
+def _configured_transport_native(sm, cp, session, now_ns, native=None, *, require_longitudinal=True, allow_restricted_lateral=False):
   """Validate the exact live native configuration and every Panda."""
   try:
     validated_here = native is None
     native = native or current_native(sm, cp, now_ns=now_ns, axis_session_id=session)
     if (not session or native is None or str(native.axisSessionId) != session or
-        native.requestedLateral and not native.lateralAllowed or
+        native.requestedLateral and not native.lateralAllowed and not allow_restricted_lateral or
         require_longitudinal and native.requestedLongitudinal and not native.longitudinalAllowed):
       return None
     coherent = bool(native.pandaInventory)
@@ -908,6 +908,35 @@ class TransportPauseFeedback:
     self.awaiting = True
     return True
 
+  def _angle_stop_intent_current(self, sm, cp, cs, axis, stamp, now_ns, *, latched, producer):
+    from opendbc.car.structs import car
+    from opendbc.car.hyundai.canfd_angle_aol import qualified, temporary_restriction
+
+    if (not qualified(cp, marked_only=True) or cs.gearShifter != car.CarState.GearShifter.drive or
+        not temporary_restriction(cp, cs) or cs.steerFaultTemporary or cs.steerFaultPermanent or
+        not latched or self.failed or self.paused or self.pending_event or self.retention_started_ns or
+        not self.lateral_only_baseline or not self.lateral_inventory or not self.lateral_baseline_ns or
+        not self.session or axis.sessionId != self.session or str(axis.faultReason) != 'none' or
+        not producer or producer != self.availability_producer or
+        axis.intentProducerSessionId != producer or axis.intentIntegrityEpoch != self.availability_epoch or
+        not axis.desiredLateral or not axis.nativeAcknowledged or
+        axis.desiredLongitudinal or axis.longitudinalActive or
+        not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
+        now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS):
+      return False
+    event_ns = int(sm.logMonoTime['onroadEvents'])
+    if (not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
+        not 0 < event_ns <= now_ns or now_ns - event_ns > 1_500_000_000 or
+        any(event.immediateDisable or event.softDisable or event.userDisable and
+            event.name not in (log.OnroadEvent.EventName.pcmDisable, log.OnroadEvent.EventName.pedalPressed)
+            for event in sm['onroadEvents'])):
+      return False
+    native = _configured_transport_native(sm, cp, self.session, now_ns, allow_restricted_lateral=True)
+    return bool(native is not None and native.pandaInventory and native.requestedLateral and
+      not native.lateralAllowed and not native.requestedLongitudinal and
+      native.observedMonoTime >= self.lateral_native_ns and
+      (native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory)) == self.lateral_inventory)
+
   def observe(self, sm, cp, cs, now_ns, *, latched=True, cancel_source_ns=0, brake_source_ns=0,
               gm_withdrawal_source_ns=0, intent_producer_session='', main_arm_ns=0):
     self.temporary_host_qualified = False
@@ -941,6 +970,10 @@ class TransportPauseFeedback:
           # the new request nor renews any command, native or healthy lease.
           self.sequence, self.signature = int(axis.sequence), signature
           return False
+      if current and continuity and self._angle_stop_intent_current(sm, cp, cs, axis, stamp, now_ns,
+          latched=latched, producer=intent_producer_session):
+        self.sequence, self.signature = int(axis.sequence), signature
+        return False
       if current and continuity and (reason == 'critical' or
           getattr(sm, 'updated', {}).get('onroadEvents', False)) and self._temporary_lag_intent_current(sm, cp, cs, axis, stamp, now_ns,
           latched=latched, producer=intent_producer_session, signature=signature):

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "28c9d821cce63586b434f23ee52d427e27a4357a8b42dd53cecb22288f420032", "schema_version": 1}
+{"protected_inventory_sha256": "6cc593e0932d0aaf3e27b098a0f5bf06484e0a9009e58686958839bcd7887064", "schema_version": 1}

tools/ci/run_vehicle_tests.py

@@ -33,6 +33,8 @@ DEBUG_SAFETY_CLASS_TARGETS = {
 
 
 RELEASE_TARGETS = (
+  "openpilot/starpilot/aol/tests/test_angle_stop_native_feedback.py",
+  "openpilot/starpilot/aol/tests/test_angle_stop_fault_guards.py",
   "openpilot/starpilot/aol/tests/test_torque_stock_cruise_transition.py",
   "openpilot/starpilot/aol/tests/test_torque_stock_brake_guards.py",
   "opendbc_repo/opendbc/safety/tests/test_hyundai_torque_stock_transition.py",