Retain fresh HKG AOL arming through stopped-engine EPS pauses

openpilot/selfdrive/car/card.py

@@ -642,7 +642,8 @@ class Car:
         brake_source_ns=brake_source_ns,
         gm_withdrawal_source_ns=gm_withdrawal_source_ns,
         intent_producer_session=getattr(self, 'slc_producer_session', ''),
-        main_arm_ns=getattr(self.aol_card_intent, 'main_arm_ns', 0),
+        main_arm_ns=(getattr(self.aol_card_intent, 'main_arm_ns', 0) or
+          getattr(self.aol_card_intent, 'physical_arm_ns', 0)),
         startup_calibration_pending=getattr(self.aol_card_intent, 'startup_calibration_pending', False),
         intent_paused=(getattr(self.aol_card_intent, 'pause_lateral', False) or
           ford_brake_pause_active(self.CP, CS, self.aol_settings)))
@@ -952,7 +953,8 @@ class Car:
         cancel_source_ns=self.aol_cancel_source(CS, now_ns), brake_source_ns=self.aol_honda_brake_source(CS, now_ns),
         gm_withdrawal_source_ns=self.aol_gm_withdrawal_source(CS, now_ns),
         intent_producer_session=getattr(self, 'slc_producer_session', ''),
-        main_arm_ns=getattr(self.aol_card_intent, 'main_arm_ns', 0),
+        main_arm_ns=(getattr(self.aol_card_intent, 'main_arm_ns', 0) or
+          getattr(self.aol_card_intent, 'physical_arm_ns', 0)),
         startup_calibration_pending=getattr(self.aol_card_intent, 'startup_calibration_pending', False),
         intent_paused=(getattr(self.aol_card_intent, 'pause_lateral', False) or
           ford_brake_pause_active(self.CP, CS, self.aol_settings))):

openpilot/starpilot/aol/tests/test_stopped_eps_arm.py

@@ -0,0 +1,103 @@
+from contextlib import ExitStack
+
+import pytest
+
+from opendbc.car import structs
+from opendbc.car.hyundai.values import CAR
+from openpilot.common.params import Params
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture
+from openpilot.starpilot.aol.tests.test_classic_scc_profiles import params
+
+
+@pytest.mark.parametrize('identity', (CAR.GENESIS_G70, CAR.HYUNDAI_SONATA, CAR.HYUNDAI_SONATA_HYBRID))
+@pytest.mark.parametrize('button', (structs.CarState.ButtonEvent.Type.mainCruise, structs.CarState.ButtonEvent.Type.lkas))
+def test_fresh_stopped_eps_press_retains_intent_until_real_permission(identity, button):
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    cleanup.enter_context(OpenpilotPrefix())
+    Params().put('MainCruiseButtonControl', 9, block=True)
+    Params().put('LKASButtonControl', 9, block=True)
+    sd, _, drive, card, receive = case._ioniq_availability_callers(
+      full_op=False, initially_armed=False, cp=params(identity, 0x391), cold_start=True)
+    def cycle(tick, *, fault=True, event=None, requested=False, allowed=False, moving=False, cruise=False):
+      state = fixture.car_state(brake=fault)
+      state.vEgo = state.vEgoRaw = 8. if moving else 0.
+      state.standstill = not moving
+      state.steerFaultTemporary = fault
+      state.cruiseState.enabled = cruise
+      if event is not None:
+        state.buttonEvents = [structs.CarState.ButtonEvent(type=button, pressed=event)]
+      axis, _, command = drive(tick, observed_state=state, native_request_lat=requested,
+        native_request_long=False, panda_controls_allowed=cruise, allowed=allowed)
+      receive(10_000_000_000 + tick * 10_000_000,
+        axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'], state=state)
+      return axis, command
+    cycle(0, fault=False)
+    assert not card.aol_card_intent.allowed_latch
+    cycle(1, event=True)
+    assert card.aol_card_intent.allowed_latch
+    for tick in range(2, 205):
+      axis, command = cycle(tick, event=False if tick == 2 else None)
+      assert card.aol_card_intent.allowed_latch, 'fresh physical EPS-wait intent was erased'
+      assert not axis.lateralActive and not command.latActive and not command.longActive
+    for tick in (205, 206):
+      axis, command = cycle(tick, fault=False)
+      assert card.aol_card_intent.allowed_latch
+      assert not axis.lateralActive and not command.latActive
+    axis, command = cycle(207, fault=False, requested=True, allowed=True, moving=True)
+    assert axis.lateralActive and command.latActive
+    assert card.aol_card_intent.allowed_latch
+    axis, command = cycle(208, fault=False, requested=True, allowed=True, moving=True, cruise=True)
+    assert sd.enabled and axis.lateralActive and command.latActive
+    assert card.aol_card_intent.allowed_latch
+    cycle(209, fault=False, event=True, requested=True, allowed=True, moving=True)
+    assert not card.aol_card_intent.allowed_latch
+
+
+@pytest.mark.parametrize('failure', ('can_invalid', 'can_timeout', 'permanent_eps', 'acc_fault',
+                                    'rx_checks', 'relay_fault', 'native_invalid', 'critical_event'))
+def test_pending_stopped_eps_arm_still_clears_for_real_faults(failure):
+  from openpilot.cereal import log
+  case = fixture.IoniqAvailabilityTests(methodName='runTest')
+  with ExitStack() as cleanup:
+    cleanup.callback(case.doCleanups)
+    cleanup.enter_context(OpenpilotPrefix())
+    Params().put('MainCruiseButtonControl', 9, block=True)
+    sd, _, drive, card, receive = case._ioniq_availability_callers(
+      full_op=False, initially_armed=False, cp=params(CAR.GENESIS_G70, 0x391), cold_start=True)
+    for tick in range(4):
+      state = fixture.car_state(brake=tick > 0)
+      state.vEgo = state.vEgoRaw = 0.
+      state.standstill = True
+      state.steerFaultTemporary = tick > 0
+      if tick in (1, 2):
+        state.buttonEvents = [structs.CarState.ButtonEvent(
+          type=structs.CarState.ButtonEvent.Type.mainCruise, pressed=tick == 1)]
+      options = {}
+      if tick == 3:
+        if failure == 'can_invalid':
+          state.canValid = False
+        elif failure == 'can_timeout':
+          state.canTimeout = True
+        elif failure == 'permanent_eps':
+          state.steerFaultPermanent = True
+        elif failure == 'acc_fault':
+          state.accFaulted = True
+        elif failure == 'rx_checks':
+          options['panda_fault'] = True
+        elif failure == 'relay_fault':
+          options['relay_fault'] = True
+        elif failure == 'native_invalid':
+          options['native_valid'] = False
+        else:
+          options['critical_event'] = log.OnroadEvent.EventName.controlsMismatch
+      axis, _, command = drive(tick, observed_state=state, native_request_lat=False,
+        native_request_long=False, allowed=False, **options)
+      receive(10_000_000_000 + tick * 10_000_000,
+        axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'], state=state)
+      assert not axis.lateralActive and not command.latActive and not command.longActive
+      if tick in (1, 2):
+        assert card.aol_card_intent.allowed_latch
+    assert not card.aol_card_intent.allowed_latch

openpilot/starpilot/aol/transport_pause.py

@@ -1395,6 +1395,13 @@ class TransportPauseFeedback:
           self.session = str(axis.sessionId)
           self.sequence, self.signature = int(axis.sequence), signature
           return False
+      if cp.brand == 'hyundai' and current and continuity:
+        from openpilot.starpilot.car.hyundai.aol import retain_stopped_eps_arm
+        if retain_stopped_eps_arm(self, sm, cp, cs, axis, now_ns, latched=latched,
+            arm_ns=main_arm_ns, producer=intent_producer_session):
+          self.session = str(axis.sessionId)
+          self.sequence, self.signature = int(axis.sequence), signature
+          return False
       if current and continuity and 0 < main_arm_ns <= now_ns <= main_arm_ns + AXIS_MAX_AGE_NS:
         from openpilot.starpilot.car.gm.aol import prior_main_off_intent_current
         if prior_main_off_intent_current(self, sm, cp, cs, axis, now_ns,

openpilot/starpilot/car/hyundai/aol.py

@@ -554,3 +554,57 @@ def pending_stock_angle_arm(CP, CS, sm, owner, *, now_ns, producer, acknowledged
                 not native.requestedLongitudinal and not native.longitudinalAllowed)
   except (AttributeError, KeyError, IndexError, TypeError, ValueError, OverflowError):
     return False
+
+
+def retain_stopped_eps_arm(feedback, sm, cp, cs, axis, now_ns, *, latched, arm_ns, producer):
+  from openpilot.cereal import log
+  from openpilot.starpilot.aol.runtime import AXIS_MAX_AGE_NS
+  from openpilot.starpilot.aol.transport_pause import _configured_transport_native
+
+  pending = getattr(feedback, '_stopped_eps_arm', None)
+  feedback._stopped_eps_arm = None
+  if not cs.steerFaultTemporary and pending is None:
+    return False
+  if (not latched or feedback.failed or feedback.paused or feedback.pending_event or
+      feedback.retention_started_ns or cs.steerFaultPermanent or not cs.canValid or cs.canTimeout or
+      cs.accFaulted or cs.doorOpen or cs.vehicleSensorsInvalid or cs.steeringDisengage or
+      cs.gearShifter not in (car.CarState.GearShifter.drive, car.CarState.GearShifter.sport,
+                            car.CarState.GearShifter.low, car.CarState.GearShifter.brake) or
+      str(axis.faultReason) != 'none' or axis.longitudinalActive or axis.desiredLongitudinal or
+      not producer or axis.intentProducerSessionId != producer or
+      not 0 < axis.sourceCarStateMonoTime <= axis.observedMonoTime <= now_ns or
+      now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS):
+    return False
+  binding = (int(arm_ns), str(axis.sessionId), producer, int(axis.intentIntegrityEpoch))
+  if pending is None:
+    if (not cs.steerFaultTemporary or not cs.standstill or not cs.brakePressed or
+        not 0 < arm_ns <= now_ns <= arm_ns + AXIS_MAX_AGE_NS or
+        not qualified_classic_scc(cp, marked_only=True) or
+        axis.lateralActive or axis.desiredLateral):
+      return False
+  elif pending[:4] != binding or not 0 <= now_ns - pending[5] <= AXIS_MAX_AGE_NS:
+    return False
+  event_ns = int(sm.logMonoTime['onroadEvents'])
+  temporary = (log.OnroadEvent.EventName.steerTempUnavailable,
+               log.OnroadEvent.EventName.steerTempUnavailableSilent)
+  allowed = (*temporary, log.OnroadEvent.EventName.wrongCarMode,
+             log.OnroadEvent.EventName.preEnableStandstill, log.OnroadEvent.EventName.pcmDisable,
+             log.OnroadEvent.EventName.pedalPressed, log.OnroadEvent.EventName.gasPressedOverride,
+             log.OnroadEvent.EventName.steerOverride)
+  events = sm['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(e.name not in allowed or e.immediateDisable or e.softDisable and e.name not in temporary for e in events) or
+      cs.steerFaultTemporary and not any(e.name in temporary for e in events)):
+    return False
+  native = _configured_transport_native(sm, cp, str(axis.sessionId), now_ns,
+    require_longitudinal=False, allow_restricted_lateral=True)
+  if native is None or not native.pandaInventory or native.requestedLongitudinal or native.longitudinalAllowed:
+    return False
+  inventory = (native.pandaSerial, tuple(slot.hardwareSerial for slot in native.pandaInventory))
+  if pending is not None and inventory != pending[4]:
+    return False
+  if axis.lateralActive:
+    return False
+  feedback._stopped_eps_arm = (*binding, inventory, now_ns)
+  return True

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "8217ee78dde766571a47c49219cafb5054495760c10e2d557e8e22bc7bd89528", "schema_version": 1}
+{"protected_inventory_sha256": "821c86f828fe4016bf78135cffa8bfb6970b6cc9fca6112ac6e2768d298961b7", "schema_version": 1}