mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-11-04 19:43:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

Normalize continuation indentation within source size limits

Browse Source
This commit is contained in:
Dom
2026-10-29 16:26:33 -05:00
parent 47e4855580
commit 3d7df63099
3 changed files with 350 additions and 350 deletions
Download Patch File Download Diff File Expand all files Collapse all files
openpilot/starpilot/aol/transport_pause.py
+309 -309
View File
@@ -7,9 +7,9 @@ from openpilot.starpilot.aol.wire import SafetyState, decode_intent
def cp_identity(cp):
return (str(cp.brand), str(cp.carFingerprint), int(cp.flags), int(cp.alternativeExperience),
bool(cp.passive), bool(cp.dashcamOnly), bool(cp.notCar),
bool(cp.openpilotLongitudinalControl), bool(cp.pcmCruise),
tuple((int(c.safetyModel.raw), int(c.safetyParam)) for c in cp.safetyConfigs))
bool(cp.passive), bool(cp.dashcamOnly), bool(cp.notCar),
bool(cp.openpilotLongitudinalControl), bool(cp.pcmCruise),
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, allow_restricted_lateral=False):
@@ -23,16 +23,16 @@ def _configured_transport_native(sm, cp, session, now_ns, native=None, *, requir
return None
coherent = bool(native.pandaInventory)
if coherent and (int(sm.logMonoTime['aolSafetyWire']) != native.observedMonoTime or
not 0 < native.observedMonoTime <= now_ns <= native.validUntilMonoTime or
now_ns - native.observedMonoTime > SAFETY_MAX_AGE_NS or
native.validUntilMonoTime - native.observedMonoTime > SAFETY_MAX_AGE_NS):
not 0 < native.observedMonoTime <= now_ns <= native.validUntilMonoTime or
now_ns - native.observedMonoTime > SAFETY_MAX_AGE_NS or
native.validUntilMonoTime - native.observedMonoTime > SAFETY_MAX_AGE_NS):
return None
# Inventory checked; validate supplied-native structure.
if coherent and (not validated_here and not native_inventory_matches_cp(native, cp) or
any(p.safetyRxChecksInvalid or p.heartbeatLost or blocking_fault_mask(p.faults, cp) or
(require_longitudinal and native.requestedLongitudinal and str(config.safetyModel) not in ('silent', 'noOutput') and
not p.controlsAllowed)
for p, config in zip(native.pandaInventory, cp.safetyConfigs, strict=True))):
any(p.safetyRxChecksInvalid or p.heartbeatLost or blocking_fault_mask(p.faults, cp) or
(require_longitudinal and native.requestedLongitudinal and str(config.safetyModel) not in ('silent', 'noOutput') and
not p.controlsAllowed)
for p, config in zip(native.pandaInventory, cp.safetyConfigs, strict=True))):
return None
# v2: exact producer-read health, monotonic 200ms lease; separate conflated BOOTTIME PandaStates: latest veto.
stamp = int(sm.logMonoTime['pandaStates'])
@@ -43,10 +43,10 @@ def _configured_transport_native(sm, cp, session, now_ns, native=None, *, requir
if not cp.safetyConfigs or len(pandas) != len(cp.safetyConfigs):
return None
return native if all(int(p.safetyModel.raw) == int(c.safetyModel.raw) and p.safetyParam == c.safetyParam and
p.alternativeExperience == cp.alternativeExperience and not p.safetyRxChecksInvalid and
not p.heartbeatLost and not blocking_fault_list(p.faults, cp) and
(not require_longitudinal or not native.requestedLongitudinal or str(c.safetyModel) in ('silent', 'noOutput') or p.controlsAllowed)
for p, c in zip(pandas, cp.safetyConfigs, strict=True)) else None
p.alternativeExperience == cp.alternativeExperience and not p.safetyRxChecksInvalid and
not p.heartbeatLost and not blocking_fault_list(p.faults, cp) and
(not require_longitudinal or not native.requestedLongitudinal or str(c.safetyModel) in ('silent', 'noOutput') or p.controlsAllowed)
for p, c in zip(pandas, cp.safetyConfigs, strict=True)) else None
except (AttributeError, KeyError, IndexError, TypeError, ValueError):
return None
@@ -57,7 +57,7 @@ def _healthy_transport_native(sm, cp, cs, session, now_ns, native=None, *, requi
if not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent:
return None
return _configured_transport_native(sm, cp, session, now_ns, native,
require_longitudinal=require_longitudinal)
require_longitudinal=require_longitudinal)
except (AttributeError, KeyError, IndexError, TypeError, ValueError):
return None
@@ -95,7 +95,7 @@ AVAILABILITY_EVENTS = (log.OnroadEvent.EventName.commIssue, log.OnroadEvent.Even
def availability_only(events):
faults = [event for event in events if event.immediateDisable or event.softDisable]
return bool(faults and all(event.name in AVAILABILITY_EVENTS and event.softDisable and
not event.immediateDisable for event in faults))
not event.immediateDisable for event in faults))
HOST_SOURCE_AVAILABILITY_EVENTS = (*AVAILABILITY_EVENTS, log.OnroadEvent.EventName.selfdrivedLagging)
@@ -105,8 +105,8 @@ def host_source_availability_only(events):
"""Closed host SOFT set for qualified inactive source-zero intent only."""
faults = [event for event in events if event.immediateDisable or event.softDisable]
return bool(faults and not any(event.userDisable for event in events) and
all(event.name in HOST_SOURCE_AVAILABILITY_EVENTS and event.softDisable and
not event.immediateDisable for event in faults))
all(event.name in HOST_SOURCE_AVAILABILITY_EVENTS and event.softDisable and
not event.immediateDisable for event in faults))
def stock_lateral_user_event(cp):
@@ -120,7 +120,7 @@ def stock_lateral_user_event(cp):
def stock_lateral_off(cp, cs, native, *, allow_independent_cancel=False, allow_independent_brake=False,
allow_independent_gas=False):
allow_independent_gas=False):
from openpilot.starpilot.nostalgia import physical_cancel
from opendbc.car.hyundai.canfd_angle_aol import qualified as angle_qualified
@@ -178,13 +178,13 @@ class AvailabilityPause:
sm_intent.settingsQualified and sm_intent.allowedLatch and sm_intent.lateralArmed and
sm_intent.producerSessionId == intent.producerSessionId)
if (raw is None or not raw.settingsQualified or not raw.allowedLatch or not raw.lateralArmed or
raw.producerSessionId != intent.producerSessionId or
not 0 < intent.carStateLogMonoTime <= now_ns or
now_ns - intent.carStateLogMonoTime > SAFETY_MAX_AGE_NS or
not 0 < raw.carStateLogMonoTime <= now_ns or raw.observedMonoTime > now_ns or
now_ns - raw.carStateLogMonoTime > SAFETY_MAX_AGE_NS or
(int(sm.logMonoTime['aolIntentWire']) != raw.carStateLogMonoTime and not projection) or
not sm.seen['aolIntentWire'] or not sm.valid['aolIntentWire'] or not sm.alive['aolIntentWire']):
raw.producerSessionId != intent.producerSessionId or
not 0 < intent.carStateLogMonoTime <= now_ns or
now_ns - intent.carStateLogMonoTime > SAFETY_MAX_AGE_NS or
not 0 < raw.carStateLogMonoTime <= now_ns or raw.observedMonoTime > now_ns or
now_ns - raw.carStateLogMonoTime > SAFETY_MAX_AGE_NS or
(int(sm.logMonoTime['aolIntentWire']) != raw.carStateLogMonoTime and not projection) or
not sm.seen['aolIntentWire'] or not sm.valid['aolIntentWire'] or not sm.alive['aolIntentWire']):
self.active = False
return False
inventory = (healthy.pandaSerial, tuple(p.hardwareSerial for p in healthy.pandaInventory))
@@ -237,12 +237,12 @@ class TransportPause:
previous = previous_intent
producer = self.producer if self.active else str(previous.producerSessionId) if previous is not None else ''
eligible = bool(eligible and latest is not None and latest[1] and raw is not None and
sm.valid['aolIntentWire'] and sm.alive['aolIntentWire'] and
raw.settingsQualified and raw.producerSessionId == producer and
latest[2].producerSessionId == producer and
int(sm.logMonoTime['aolIntentWire']) == raw.carStateLogMonoTime and
0 < raw.carStateLogMonoTime <= now_ns and raw.observedMonoTime <= now_ns and
(self.epoch is None or self.epoch == companion.integrity_epoch))
sm.valid['aolIntentWire'] and sm.alive['aolIntentWire'] and
raw.settingsQualified and raw.producerSessionId == producer and
latest[2].producerSessionId == producer and
int(sm.logMonoTime['aolIntentWire']) == raw.carStateLogMonoTime and
0 < raw.carStateLogMonoTime <= now_ns and raw.observedMonoTime <= now_ns and
(self.epoch is None or self.epoch == companion.integrity_epoch))
if self.active:
if not eligible:
self.active = False
@@ -317,23 +317,23 @@ class TransportPauseFeedback:
cp.openpilotLongitudinalControl and cp.carFingerprint == CAR.FORD_MUSTANG_MACH_E_MK1)
pending_ack = str(axis.mode) == 'off' and not axis.lateralActive
if (not first_long or not latched or self.failed or self.paused or
self.pending_event or self.retention_started_ns or self.source_retention_started_ns or
self.lateral_baseline_ns or not self.session or str(axis.sessionId) != self.session or
not producer or str(axis.intentProducerSessionId) != producer or int(axis.intentIntegrityEpoch) < 0 or
self.signature is None or self.signature[14:16] != (producer, int(axis.intentIntegrityEpoch)) or
str(axis.faultReason) != 'none' or not (pending_ack or str(axis.mode) == 'lateralOnly' and axis.lateralActive) or
not axis.nativeAcknowledged or not axis.desiredLateral or not axis.desiredLongitudinal or axis.longitudinalActive or
not 0 < axis.sourceCarStateMonoTime <= axis.observedMonoTime <= now_ns <= axis.validUntilMonoTime or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
not cs.gasPressed or cs.brakePressed or cs.regenBraking or not cs.cruiseState.available or
not cs.cruiseState.enabled or not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or
cs.steerFaultPermanent or cs.accFaulted or cs.vehicleSensorsInvalid or cs.steeringDisengage or
cs.doorOpen or cs.seatbeltUnlatched or cs.gearShifter not in
(structs.CarState.GearShifter.drive, structs.CarState.GearShifter.sport, structs.CarState.GearShifter.low) or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
not 0 < int(sm.logMonoTime['onroadEvents']) <= now_ns or
now_ns - int(sm.logMonoTime['onroadEvents']) > 1_500_000_000 or
any(event.immediateDisable or event.softDisable or event.userDisable for event in sm['onroadEvents'])):
self.pending_event or self.retention_started_ns or self.source_retention_started_ns or
self.lateral_baseline_ns or not self.session or str(axis.sessionId) != self.session or
not producer or str(axis.intentProducerSessionId) != producer or int(axis.intentIntegrityEpoch) < 0 or
self.signature is None or self.signature[14:16] != (producer, int(axis.intentIntegrityEpoch)) or
str(axis.faultReason) != 'none' or not (pending_ack or str(axis.mode) == 'lateralOnly' and axis.lateralActive) or
not axis.nativeAcknowledged or not axis.desiredLateral or not axis.desiredLongitudinal or axis.longitudinalActive or
not 0 < axis.sourceCarStateMonoTime <= axis.observedMonoTime <= now_ns <= axis.validUntilMonoTime or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
not cs.gasPressed or cs.brakePressed or cs.regenBraking or not cs.cruiseState.available or
not cs.cruiseState.enabled or not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or
cs.steerFaultPermanent or cs.accFaulted or cs.vehicleSensorsInvalid or cs.steeringDisengage or
cs.doorOpen or cs.seatbeltUnlatched or cs.gearShifter not in
(structs.CarState.GearShifter.drive, structs.CarState.GearShifter.sport, structs.CarState.GearShifter.low) or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
not 0 < int(sm.logMonoTime['onroadEvents']) <= now_ns or
now_ns - int(sm.logMonoTime['onroadEvents']) > 1_500_000_000 or
any(event.immediateDisable or event.softDisable or event.userDisable for event in sm['onroadEvents'])):
return None
native = independent_lateral_health(sm, cp, cs, self.session, now_ns)
# Pre-ACK newer actual 10Hz LAT grant: intent only, never denied LONG.
@@ -376,26 +376,26 @@ class TransportPauseFeedback:
cs.gearShifter in (structs.CarState.GearShifter.drive, structs.CarState.GearShifter.sport,
structs.CarState.GearShifter.low, structs.CarState.GearShifter.brake))
if (not latched or self.failed or self.paused or self.pending_event or self.retention_started_ns or
not self.session or str(axis.sessionId) != self.session or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
not self.lateral_inventory or not self.lateral_baseline_ns or
str(axis.faultReason) != 'none' or not early_long and str(axis.mode) != 'lateralOnly' or not axis.nativeAcknowledged or
not axis.desiredLateral or not axis.lateralActive or axis.longitudinalActive and not early_long or
not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= axis.observedMonoTime <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or cs.accFaulted or
cs.steeringDisengage or physical_cancel(cs, cp) and not ford_withdrawal or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
not 0 < int(sm.logMonoTime['onroadEvents']) <= now_ns or
now_ns - int(sm.logMonoTime['onroadEvents']) > 1_500_000_000 or
any(event.immediateDisable or event.softDisable or
event.userDisable and not (ev9_main_off and event.name == log.OnroadEvent.EventName.wrongCarMode or
(gm_pedal or classic_brake_withdrawal(cp, cs)) and event.name == log.OnroadEvent.EventName.pedalPressed or
ford_withdrawal and event.name in (log.OnroadEvent.EventName.pedalPressed, log.OnroadEvent.EventName.pcmDisable,
log.OnroadEvent.EventName.buttonCancel))
for event in sm['onroadEvents'])):
not self.session or str(axis.sessionId) != self.session or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
not self.lateral_inventory or not self.lateral_baseline_ns or
str(axis.faultReason) != 'none' or not early_long and str(axis.mode) != 'lateralOnly' or not axis.nativeAcknowledged or
not axis.desiredLateral or not axis.lateralActive or axis.longitudinalActive and not early_long or
not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= axis.observedMonoTime <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or cs.accFaulted or
cs.steeringDisengage or physical_cancel(cs, cp) and not ford_withdrawal or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
not 0 < int(sm.logMonoTime['onroadEvents']) <= now_ns or
now_ns - int(sm.logMonoTime['onroadEvents']) > 1_500_000_000 or
any(event.immediateDisable or event.softDisable or
event.userDisable and not (ev9_main_off and event.name == log.OnroadEvent.EventName.wrongCarMode or
(gm_pedal or classic_brake_withdrawal(cp, cs)) and event.name == log.OnroadEvent.EventName.pedalPressed or
ford_withdrawal and event.name in (log.OnroadEvent.EventName.pedalPressed, log.OnroadEvent.EventName.pcmDisable,
log.OnroadEvent.EventName.buttonCancel))
for event in sm['onroadEvents'])):
return None
native = independent_lateral_health(sm, cp, cs, self.session, now_ns)
if (native is None or not native.pandaInventory or not native.requestedLongitudinal or native.longitudinalAllowed or
@@ -405,7 +405,7 @@ class TransportPauseFeedback:
owners = [index for index, config in enumerate(cp.safetyConfigs)
if str(config.safetyModel) not in ('silent', 'noOutput')]
if not (len(owners) == 1 and (early_long or not native.pandaInventory[owners[0]].controlsAllowed and
not sm['pandaStates'][owners[0]].controlsAllowed)):
not sm['pandaStates'][owners[0]].controlsAllowed)):
return None
return native
@@ -418,20 +418,20 @@ class TransportPauseFeedback:
pending_baseline = self.ordinary_warning_pending_baseline
pending = bool(pending_baseline is not None and self.pending_event)
if (not latched or self.failed or self.paused or self.pending_event and not pending or self.retention_started_ns or
not self.session or not self.lateral_baseline_ns or not self.lateral_inventory or
not (policy := policy_for(cp)).intent_supported or
policy.cruise_main_required and not cs.cruiseState.available or
str(axis.sessionId) != self.session or
not producer or producer != self.availability_producer or axis.intentProducerSessionId != producer or
axis.intentIntegrityEpoch != self.availability_epoch or
not cs.canValid or cs.canTimeout or cs.steerFaultPermanent or cs.accFaulted or cs.doorOpen or
cs.steeringDisengage or
cs.gearShifter not in (structs.CarState.GearShifter.drive, structs.CarState.GearShifter.sport,
structs.CarState.GearShifter.low, structs.CarState.GearShifter.brake) or
not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
str(axis.faultReason) not in ('none', 'availabilityPause') or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents']):
not self.session or not self.lateral_baseline_ns or not self.lateral_inventory or
not (policy := policy_for(cp)).intent_supported or
policy.cruise_main_required and not cs.cruiseState.available or
str(axis.sessionId) != self.session or
not producer or producer != self.availability_producer or axis.intentProducerSessionId != producer or
axis.intentIntegrityEpoch != self.availability_epoch or
not cs.canValid or cs.canTimeout or cs.steerFaultPermanent or cs.accFaulted or cs.doorOpen or
cs.steeringDisengage or
cs.gearShifter not in (structs.CarState.GearShifter.drive, structs.CarState.GearShifter.sport,
structs.CarState.GearShifter.low, structs.CarState.GearShifter.brake) or
not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
str(axis.faultReason) not in ('none', 'availabilityPause') or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents']):
return False
event_ns = int(sm.logMonoTime['onroadEvents'])
events = sm['onroadEvents']
@@ -523,7 +523,7 @@ class TransportPauseFeedback:
all(event.softDisable and not event.immediateDisable and event.name in
(log.OnroadEvent.EventName.locationdTemporaryError, log.OnroadEvent.EventName.commIssue) for event in faults))
temporary = location_fault or bool(len(faults) == 1 and faults[0].name in names and
faults[0].softDisable and not faults[0].immediateDisable)
faults[0].softDisable and not faults[0].immediateDisable)
eps_recovery = bool(not cs.steerFaultTemporary and getattr(self, 'temporary_eps_event_ns', 0) and
axis.faultEventMonoTime == self.temporary_eps_event_ns and
all(not event.immediateDisable and
@@ -577,16 +577,16 @@ class TransportPauseFeedback:
if axis.longitudinalActive and not location_fault or not (temporary or eps_recovery or pedal):
return False
if (not latched or self.failed or self.paused or self.pending_event or self.retention_started_ns or
not self.session or str(axis.sessionId) != self.session or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
not self.lateral_inventory or not self.lateral_baseline_ns or
not cs.canValid or cs.canTimeout or cs.steerFaultPermanent or cs.accFaulted or cs.steerFaultTemporary or
users and not (pedal or stock_temporary_user) or str(axis.faultReason) not in ('none', 'critical') or
not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents']):
not self.session or str(axis.sessionId) != self.session or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
not self.lateral_inventory or not self.lateral_baseline_ns or
not cs.canValid or cs.canTimeout or cs.steerFaultPermanent or cs.accFaulted or cs.steerFaultTemporary or
users and not (pedal or stock_temporary_user) or str(axis.faultReason) not in ('none', 'critical') or
not 0 < self.lateral_source_ns <= axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents']):
return False
event_ns = int(sm.logMonoTime['onroadEvents'])
if not 0 < event_ns <= now_ns or now_ns - event_ns > 1_500_000_000:
@@ -598,11 +598,11 @@ class TransportPauseFeedback:
temporary and axis.desiredLateral and axis.lateralActive and axis.nativeAcknowledged and
0 < axis.faultEventMonoTime < event_ns and now_ns - event_ns <= AXIS_MAX_AGE_NS)
if not (event_first or temporary and axis.faultEventMonoTime == event_ns and
(withdrawn or str(axis.faultReason) == 'none' and axis.desiredLateral and
axis.lateralActive and axis.nativeAcknowledged) or eps_recovery and withdrawn or
pedal and str(axis.faultReason) == 'none' and axis.desiredLateral and axis.lateralActive and
not axis.desiredLongitudinal and not axis.longitudinalActive and
axis.nativeAcknowledged and axis.faultEventMonoTime == event_ns):
(withdrawn or str(axis.faultReason) == 'none' and axis.desiredLateral and
axis.lateralActive and axis.nativeAcknowledged) or eps_recovery and withdrawn or
pedal and str(axis.faultReason) == 'none' and axis.desiredLateral and axis.lateralActive and
not axis.desiredLongitudinal and not axis.longitudinalActive and
axis.nativeAcknowledged and axis.faultEventMonoTime == event_ns):
return False
native = _configured_transport_native(sm, cp, self.session, now_ns, require_longitudinal=not pedal)
if (native is None or not native.pandaInventory or native.observedMonoTime < self.lateral_native_ns or
@@ -627,7 +627,7 @@ class TransportPauseFeedback:
events = sm['onroadEvents']
faults = [event for event in events if event.immediateDisable or event.softDisable]
isolated = bool(len(faults) == 1 and faults[0].name == log.OnroadEvent.EventName.selfdrivedLagging and
faults[0].softDisable and not faults[0].immediateDisable)
faults[0].softDisable and not faults[0].immediateDisable)
# Healthy skips policy/native/provenance. Pre-event critical proof grants no lag exemption.
recovering = bool(cp.brand == 'gm' and not faults and reason == 'none' and
0 < self.lag_withdrawal_ns <= now_ns <= self.lag_withdrawal_ns + SAFETY_MAX_AGE_NS and
@@ -639,14 +639,14 @@ class TransportPauseFeedback:
joining = bool(self.pending_event and previous is not None and previous[5] == 'critical' and
previous[6] == self.pending_event)
if (not latched or self.failed or self.paused or self.pending_event and not joining or self.retention_started_ns or
not self.session or str(axis.sessionId) != self.session or reason not in ('none', 'critical') or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
not self.lateral_baseline_ns or not self.lateral_inventory or
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or cs.accFaulted or
not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS):
not self.session or str(axis.sessionId) != self.session or reason not in ('none', 'critical') or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
not self.lateral_baseline_ns or not self.lateral_inventory or
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or cs.accFaulted 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
@@ -655,7 +655,7 @@ class TransportPauseFeedback:
(axis.faultEventMonoTime != self.pending_event or now_ns - self.pending_event > AXIS_MAX_AGE_NS)):
return False
withdrawn = bool(reason == 'critical' and not axis.desiredLateral and not axis.desiredLongitudinal and
not axis.lateralActive and not axis.longitudinalActive)
not axis.lateralActive and not axis.longitudinalActive)
matched = bool(withdrawn and isolated and axis.faultEventMonoTime == event_ns <= stamp)
held_original = bool(reason == 'none' and isolated and signature == previous and
axis.desiredLateral and axis.lateralActive and axis.nativeAcknowledged and
@@ -718,73 +718,73 @@ class TransportPauseFeedback:
gas_was_pressed = self.temporary_eps_gas_override
previous = self.signature
if (cp.brand != 'ford' or cp.openpilotLongitudinalControl or
not cp.safetyConfigs[-1].safetyParam & 128 or
not latched or not gas_was_pressed or cs.gasPressed or
not temporary_eps_restriction(cp, cs) or not self.lateral_only_baseline or
self.failed or self.paused or self.pending_event or self.retention_started_ns or
not self.ever_lateral_active or cp_identity(cp) != self.cp or previous is None or
not self.temporary_eps_event_ns or not self.lateral_inventory or not self.lateral_baseline_ns or
previous[0] != self.session or previous[7] != self.session or previous[1] != self.sequence or
previous[5] != 'none' or any(previous[9:13]) or not previous[13] or
previous[6] != self.temporary_eps_event_ns or
previous[14] != producer or previous[15] != self.availability_epoch or
not 0 < previous[8] <= previous[2] <= now_ns <= previous[4] or
now_ns - previous[8] > AXIS_MAX_AGE_NS):
not cp.safetyConfigs[-1].safetyParam & 128 or
not latched or not gas_was_pressed or cs.gasPressed or
not temporary_eps_restriction(cp, cs) or not self.lateral_only_baseline or
self.failed or self.paused or self.pending_event or self.retention_started_ns or
not self.ever_lateral_active or cp_identity(cp) != self.cp or previous is None or
not self.temporary_eps_event_ns or not self.lateral_inventory or not self.lateral_baseline_ns or
previous[0] != self.session or previous[7] != self.session or previous[1] != self.sequence or
previous[5] != 'none' or any(previous[9:13]) or not previous[13] or
previous[6] != self.temporary_eps_event_ns or
previous[14] != producer or previous[15] != self.availability_epoch or
not 0 < previous[8] <= previous[2] <= now_ns <= previous[4] or
now_ns - previous[8] > AXIS_MAX_AGE_NS):
return False
if (not axis.qualified or str(axis.sessionId) != self.session or str(axis.faultSessionId) != self.session or
str(axis.intentProducerSessionId) != producer or int(axis.intentIntegrityEpoch) != self.availability_epoch or
int(axis.sequence) < self.sequence or str(axis.faultReason) != 'none' or str(axis.mode) != 'off' or
not axis.nativeAcknowledged or any((axis.desiredLateral,axis.lateralActive,axis.desiredLongitudinal,axis.longitudinalActive)) or
not sm.seen['aolAxisState'] or not sm.valid['aolAxisState'] or not sm.alive['aolAxisState'] or
not previous[8] <= axis.sourceCarStateMonoTime <= stamp == axis.observedMonoTime <= now_ns <= axis.validUntilMonoTime or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
axis.validUntilMonoTime - stamp > AXIS_MAX_AGE_NS):
str(axis.intentProducerSessionId) != producer or int(axis.intentIntegrityEpoch) != self.availability_epoch or
int(axis.sequence) < self.sequence or str(axis.faultReason) != 'none' or str(axis.mode) != 'off' or
not axis.nativeAcknowledged or any((axis.desiredLateral,axis.lateralActive,axis.desiredLongitudinal,axis.longitudinalActive)) or
not sm.seen['aolAxisState'] or not sm.valid['aolAxisState'] or not sm.alive['aolAxisState'] or
not previous[8] <= axis.sourceCarStateMonoTime <= stamp == axis.observedMonoTime <= now_ns <= axis.validUntilMonoTime or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
axis.validUntilMonoTime - stamp > AXIS_MAX_AGE_NS):
return False
event_ns = int(sm.logMonoTime['onroadEvents'])
events = sm['onroadEvents']
temporary_names = ('steerTempUnavailable','steerTempUnavailableSilent')
if (axis.faultEventMonoTime == event_ns or
min(axis.faultEventMonoTime,event_ns) != self.temporary_eps_event_ns or
not 0 < max(axis.faultEventMonoTime,event_ns) <= now_ns or
now_ns-max(axis.faultEventMonoTime,event_ns) > AXIS_MAX_AGE_NS or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
now_ns-event_ns > 1_500_000_000 or
not any(str(e.name)in temporary_names for e in events) or
any(e.immediateDisable or e.softDisable and str(e.name)!='steerTempUnavailable' or
e.userDisable and str(e.name)!='pcmDisable' or
str(e.name)not in (*temporary_names,'pcmDisable','gasPressedOverride','steerOverride') for e in events)):
min(axis.faultEventMonoTime,event_ns) != self.temporary_eps_event_ns or
not 0 < max(axis.faultEventMonoTime,event_ns) <= now_ns or
now_ns-max(axis.faultEventMonoTime,event_ns) > AXIS_MAX_AGE_NS or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
now_ns-event_ns > 1_500_000_000 or
not any(str(e.name)in temporary_names for e in events) or
any(e.immediateDisable or e.softDisable and str(e.name)!='steerTempUnavailable' or
e.userDisable and str(e.name)!='pcmDisable' or
str(e.name)not in (*temporary_names,'pcmDisable','gasPressedOverride','steerOverride') for e in events)):
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
not any((native.requestedLateral,native.lateralAllowed,native.requestedLongitudinal,native.longitudinalAllowed)) and
native.observedMonoTime>=self.lateral_native_ns and
self.lateral_inventory==(native.pandaSerial,tuple(p.hardwareSerial for p in native.pandaInventory)))
not any((native.requestedLateral,native.lateralAllowed,native.requestedLongitudinal,native.longitudinalAllowed)) and
native.observedMonoTime>=self.lateral_native_ns and
self.lateral_inventory==(native.pandaSerial,tuple(p.hardwareSerial for p in native.pandaInventory)))
def _temporary_steering_intent_current(self, sm, cp, cs, axis, stamp, now_ns, *, latched, producer, signature):
"""Keep an existing armed intent during isolated temporary EPS withdrawal."""
if (not latched or self.failed or self.paused or self.retention_started_ns or
not self.session or str(axis.sessionId) != self.session or not cs.steerFaultTemporary or
cs.steerFaultPermanent or not cs.canValid or cs.canTimeout or
str(axis.faultReason) != 'none' or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
not self.lateral_baseline_ns or not self.lateral_inventory or
not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS):
not self.session or str(axis.sessionId) != self.session or not cs.steerFaultTemporary or
cs.steerFaultPermanent or not cs.canValid or cs.canTimeout or
str(axis.faultReason) != 'none' or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
not self.lateral_baseline_ns or not self.lateral_inventory or
not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS):
return False
withdrawn = not axis.desiredLateral and not axis.lateralActive
previous = self.signature
pending = bool(self.pending_event)
all_off = bool(withdrawn and not axis.desiredLongitudinal and not axis.longitudinalActive and str(axis.mode) == 'off')
if pending and (not all_off or previous is None or previous[5] != 'none' or
previous[0] != self.session or previous[7] != self.session or
previous[1] != self.sequence or previous[6] != self.pending_event or
previous[9] or previous[10] or previous[11] or previous[12] or
previous[14] != producer or previous[15] != self.availability_epoch or
axis.faultEventMonoTime != self.pending_event or
axis.sourceCarStateMonoTime < previous[8] or stamp < previous[2] or
not 0 < self.pending_event <= now_ns or now_ns - self.pending_event > AXIS_MAX_AGE_NS):
previous[0] != self.session or previous[7] != self.session or
previous[1] != self.sequence or previous[6] != self.pending_event or
previous[9] or previous[10] or previous[11] or previous[12] or
previous[14] != producer or previous[15] != self.availability_epoch or
axis.faultEventMonoTime != self.pending_event or
axis.sourceCarStateMonoTime < previous[8] or stamp < previous[2] or
not 0 < self.pending_event <= now_ns or now_ns - self.pending_event > AXIS_MAX_AGE_NS):
return False
# Prior ACTIVE identity in first temporary-EPS Card cycle: original source/output lease only;
# No permission-deadline/baseline renewal.
@@ -798,7 +798,7 @@ class TransportPauseFeedback:
0 < self.lateral_source_ns <= now_ns and
now_ns - self.lateral_source_ns <= AXIS_MAX_AGE_NS)
held_original = bool(axis.desiredLateral and axis.lateralActive and axis.nativeAcknowledged and
(signature == previous or fresh_active_successor))
(signature == previous or fresh_active_successor))
if not withdrawn and not held_original:
return False
event_ns = int(sm.logMonoTime['onroadEvents'])
@@ -808,7 +808,7 @@ class TransportPauseFeedback:
not axis.longitudinalActive and event.name == log.OnroadEvent.EventName.seatbeltNotLatched and
event.softDisable and event.noEntry and not event.immediateDisable and not event.userDisable))
temporary_names = (log.OnroadEvent.EventName.steerTempUnavailable,
log.OnroadEvent.EventName.steerTempUnavailableSilent)
log.OnroadEvent.EventName.steerTempUnavailableSilent)
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 and event.name != temporary_names[0] for event in events) or
@@ -892,17 +892,17 @@ class TransportPauseFeedback:
source_ns = self.honda_brake_ns or (brake_source_ns if cs.brakePressed else 0)
if (not qualified(cp, marked_only=True) or not latched or self.failed or self.paused or self.pending_event or
not self.session or str(axis.sessionId) != self.session or
not self.healthy_feedback_ns or not 0 <= now_ns - self.healthy_feedback_ns <= AXIS_MAX_AGE_NS or
not 0 < source_ns <= now_ns or now_ns - source_ns > SAFETY_MAX_AGE_NS or
(not self.honda_brake_ns and now_ns - source_ns > AXIS_MAX_AGE_NS) or
cs.accFaulted or cs.seatbeltUnlatched 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 not axis.desiredLateral or not axis.lateralActive or
(not axis.nativeAcknowledged and (axis.desiredLongitudinal or axis.longitudinalActive)) or
not 0 < axis.sourceCarStateMonoTime <= axis.observedMonoTime or
axis.observedMonoTime - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS):
not self.session or str(axis.sessionId) != self.session or
not self.healthy_feedback_ns or not 0 <= now_ns - self.healthy_feedback_ns <= AXIS_MAX_AGE_NS or
not 0 < source_ns <= now_ns or now_ns - source_ns > SAFETY_MAX_AGE_NS or
(not self.honda_brake_ns and now_ns - source_ns > AXIS_MAX_AGE_NS) or
cs.accFaulted or cs.seatbeltUnlatched 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 not axis.desiredLateral or not axis.lateralActive or
(not axis.nativeAcknowledged and (axis.desiredLongitudinal or axis.longitudinalActive)) or
not 0 < axis.sourceCarStateMonoTime <= axis.observedMonoTime or
axis.observedMonoTime - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS):
return None
event_ns = int(sm.logMonoTime['onroadEvents'])
if (not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
@@ -933,7 +933,7 @@ class TransportPauseFeedback:
return None
cancel = any(e.type == car.CarState.ButtonEvent.Type.cancel and e.pressed for e in cs.buttonEvents)
fresh_cancel = bool(cancel and self.last_cancel_source_ns < cancel_source_ns <= now_ns and
now_ns - cancel_source_ns <= AXIS_MAX_AGE_NS)
now_ns - cancel_source_ns <= AXIS_MAX_AGE_NS)
if not self.longitudinal_cancel_ns and not fresh_cancel:
return None
if not withdrawal_required and (self.longitudinal_cancel_ns or not fresh_cancel):
@@ -959,10 +959,10 @@ class TransportPauseFeedback:
return None
# controlsAllowed: normal/LONG-off; keep latest raw config/fault vetoes + exact producer-read inventory.
current_pandas = all(int(p.safetyModel.raw) == int(c.safetyModel.raw) and p.safetyParam == c.safetyParam and
p.alternativeExperience == cp.alternativeExperience and not p.safetyRxChecksInvalid and
not p.heartbeatLost and not blocking_fault_list(p.faults, cp) and
(str(c.safetyModel) not in ('silent', 'noOutput') or not p.controlsAllowed)
for p, c in zip(pandas, cp.safetyConfigs, strict=True))
p.alternativeExperience == cp.alternativeExperience and not p.safetyRxChecksInvalid and
not p.heartbeatLost and not blocking_fault_list(p.faults, cp) and
(str(c.safetyModel) not in ('silent', 'noOutput') or not p.controlsAllowed)
for p, c in zip(pandas, cp.safetyConfigs, strict=True))
return native if current_pandas else None
@@ -972,16 +972,16 @@ class TransportPauseFeedback:
if reason not in ('none', 'availabilityPause') or reason == 'none' and not availability_only(events):
return False
if not (latched and current and continuity and not self.paused and not self.pending_event and
cp_identity(cp) == self.cp and intent_producer_session and
getattr(axis, 'intentProducerSessionId', '') == intent_producer_session == self.availability_producer and
int(getattr(axis, 'intentIntegrityEpoch', 0)) == self.availability_epoch and
self.session and
self.lateral_inventory and axis.desiredLateral and
(str(axis.faultReason) == 'none' or not axis.desiredLongitudinal) and
axis.nativeAcknowledged and (
0 < axis.sourceCarStateMonoTime <= stamp and stamp - axis.sourceCarStateMonoTime <= AXIS_MAX_AGE_NS or
reason == 'availabilityPause' and axis.sourceCarStateMonoTime == 0 and
not axis.lateralActive and not axis.longitudinalActive)):
cp_identity(cp) == self.cp and intent_producer_session and
getattr(axis, 'intentProducerSessionId', '') == intent_producer_session == self.availability_producer and
int(getattr(axis, 'intentIntegrityEpoch', 0)) == self.availability_epoch and
self.session and
self.lateral_inventory and axis.desiredLateral and
(str(axis.faultReason) == 'none' or not axis.desiredLongitudinal) and
axis.nativeAcknowledged and (
0 < axis.sourceCarStateMonoTime <= stamp and stamp - axis.sourceCarStateMonoTime <= AXIS_MAX_AGE_NS or
reason == 'availabilityPause' and axis.sourceCarStateMonoTime == 0 and
not axis.lateralActive and not axis.longitudinalActive)):
return False
native = _healthy_transport_native(sm, cp, cs, str(axis.sessionId), now_ns)
if (native is None or not native.requestedLateral or not native.lateralAllowed or
@@ -1031,9 +1031,9 @@ class TransportPauseFeedback:
event_ns = int(sm.logMonoTime['onroadEvents'])
faults = [event for event in events if event.immediateDisable or event.softDisable]
live_event = bool(sm.valid['onroadEvents'] and sm.alive['onroadEvents'] and
0 < event_ns <= now_ns and now_ns - event_ns <= 1_500_000_000)
0 < event_ns <= now_ns and now_ns - event_ns <= 1_500_000_000)
host_source_off = bool(reason == 'availabilityPause' and axis.sourceCarStateMonoTime == 0 and
not axis.lateralActive and not axis.longitudinalActive)
not axis.lateralActive and not axis.longitudinalActive)
issue = live_event and (availability_only(events) or host_source_off and host_source_availability_only(events))
if reason not in ('none', 'availabilityPause') or faults and not issue:
return False
@@ -1079,17 +1079,17 @@ class TransportPauseFeedback:
# and current native LAT grant.
verified_pause = self.retention_verified or self.ordinary_warning_retained
if (not latched or self.failed or self.paused or self.pending_event or not verified_pause or
not self.retention_started_ns or not 0 <= now_ns - self.retention_started_ns <= AXIS_MAX_AGE_NS or
not current or not continuity or str(axis.sessionId) != self.session or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
str(axis.faultReason) != 'none' or not axis.nativeAcknowledged or
not axis.desiredLateral or not axis.lateralActive or
not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
not 0 < event_ns == axis.faultEventMonoTime <= stamp):
not self.retention_started_ns or not 0 <= now_ns - self.retention_started_ns <= AXIS_MAX_AGE_NS or
not current or not continuity or str(axis.sessionId) != self.session or
not producer or producer != self.availability_producer or
str(getattr(axis, 'intentProducerSessionId', '')) != producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch or
str(axis.faultReason) != 'none' or not axis.nativeAcknowledged or
not axis.desiredLateral or not axis.lateralActive or
not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
not 0 < event_ns == axis.faultEventMonoTime <= stamp):
return False
events = sm['onroadEvents']
faults = [event for event in events if event.userDisable or event.softDisable or event.immediateDisable]
@@ -1099,13 +1099,13 @@ class TransportPauseFeedback:
# no pending-deadline/old-command-lease renewal.
previous = self.signature
if (previous is None or previous[0] != self.session or previous[5] != 'availabilityPause' or
previous[7] != self.session or previous[8] != 0 or previous[9] or previous[10] or
not previous[11] or previous[12] or not previous[13] or
previous[14] != producer or previous[15] != self.availability_epoch or
not 0 < previous[6] < event_ns or self.retention_event_ns != event_ns or
len(faults) != 1 or faults[0].name != log.OnroadEvent.EventName.selfdrivedLagging or
not faults[0].softDisable or faults[0].userDisable or faults[0].immediateDisable or
any(event.noEntry and event.name != log.OnroadEvent.EventName.selfdrivedLagging for event in events)):
previous[7] != self.session or previous[8] != 0 or previous[9] or previous[10] or
not previous[11] or previous[12] or not previous[13] or
previous[14] != producer or previous[15] != self.availability_epoch or
not 0 < previous[6] < event_ns or self.retention_event_ns != event_ns or
len(faults) != 1 or faults[0].name != log.OnroadEvent.EventName.selfdrivedLagging or
not faults[0].softDisable or faults[0].userDisable or faults[0].immediateDisable or
any(event.noEntry and event.name != log.OnroadEvent.EventName.selfdrivedLagging for event in events)):
return False
elif (axis.desiredLongitudinal or axis.longitudinalActive or
not self.retention_event_ns < event_ns or faults):
@@ -1120,13 +1120,13 @@ class TransportPauseFeedback:
"""An established combined producer advanced while Card missed its output lease."""
previous = self.signature
if (now_ns <= signature[4] or previous is None or self.retention_started_ns or
not (sm.seen['aolAxisState'] and sm.valid['aolAxisState'] and sm.alive['aolAxisState']) or
not axis.desiredLongitudinal or not axis.longitudinalActive or
not all(previous[index] for index in (9, 10, 11, 12, 13)) or previous[5] != 'none' or
signature[1] <= self.sequence or stamp <= previous[2] or
not 0 < self.lateral_source_ns <= previous[8] < signature[8] <= stamp or
stamp - signature[8] > AXIS_MAX_AGE_NS or
signature[4] != stamp + AXIS_MAX_AGE_NS):
not (sm.seen['aolAxisState'] and sm.valid['aolAxisState'] and sm.alive['aolAxisState']) or
not axis.desiredLongitudinal or not axis.longitudinalActive or
not all(previous[index] for index in (9, 10, 11, 12, 13)) or previous[5] != 'none' or
signature[1] <= self.sequence or stamp <= previous[2] or
not 0 < self.lateral_source_ns <= previous[8] < signature[8] <= stamp or
stamp - signature[8] > AXIS_MAX_AGE_NS or
signature[4] != stamp + AXIS_MAX_AGE_NS):
return False
from openpilot.starpilot.car.hyundai.aol import qualified_ioniq6
return qualified_ioniq6(cp)
@@ -1142,21 +1142,21 @@ class TransportPauseFeedback:
not any(signature[9:14]) and 0 < signature[6] <= stamp and
0 < self.lateral_source_ns <= signature[8] <= stamp and stamp - signature[8] <= AXIS_MAX_AGE_NS)
if (not latched or self.failed or self.paused or self.pending_event or not self.session or
cp_identity(cp) != self.cp or not continuity or
str(axis.sessionId) != self.session or str(axis.faultSessionId) != self.session or
not axis.qualified or (str(axis.faultReason) != 'none' or not axis.nativeAcknowledged or
not axis.desiredLateral or not axis.lateralActive) and not matched_lag or
(axis.desiredLongitudinal or axis.longitudinalActive) and not held and
not self._new_expired_combined_source(sm, cp, axis, stamp, now_ns, signature) or
(not sm.seen['aolAxisState'] or not sm.valid['aolAxisState']) and not held or
int(axis.observedMonoTime) != stamp or not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
not 0 <= axis.validUntilMonoTime - stamp <= AXIS_MAX_AGE_NS or
now_ns - axis.sourceCarStateMonoTime > SAFETY_MAX_AGE_NS or
not (stamp - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or now_ns > axis.validUntilMonoTime or
held and output_missing) or
not producer or getattr(axis, 'intentProducerSessionId', '') != producer or
producer != self.availability_producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch):
cp_identity(cp) != self.cp or not continuity or
str(axis.sessionId) != self.session or str(axis.faultSessionId) != self.session or
not axis.qualified or (str(axis.faultReason) != 'none' or not axis.nativeAcknowledged or
not axis.desiredLateral or not axis.lateralActive) and not matched_lag or
(axis.desiredLongitudinal or axis.longitudinalActive) and not held and
not self._new_expired_combined_source(sm, cp, axis, stamp, now_ns, signature) or
(not sm.seen['aolAxisState'] or not sm.valid['aolAxisState']) and not held or
int(axis.observedMonoTime) != stamp or not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
not 0 <= axis.validUntilMonoTime - stamp <= AXIS_MAX_AGE_NS or
now_ns - axis.sourceCarStateMonoTime > SAFETY_MAX_AGE_NS or
not (stamp - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or now_ns > axis.validUntilMonoTime or
held and output_missing) or
not producer or getattr(axis, 'intentProducerSessionId', '') != producer or
producer != self.availability_producer or
int(getattr(axis, 'intentIntegrityEpoch', 0)) != self.availability_epoch):
return False
if not 0 < self.lateral_source_ns <= now_ns <= self.lateral_source_ns + SAFETY_MAX_AGE_NS:
return False
@@ -1215,18 +1215,18 @@ class TransportPauseFeedback:
if now_ns <= axis.validUntilMonoTime or signature != self.signature:
return False
if (not latched or self.failed or self.paused or not self.session or native is None or
cp_identity(cp) != self.cp or
str(axis.sessionId) != self.session or
str(axis.faultSessionId) != self.session or str(axis.faultReason) != 'none' or
not axis.qualified or not axis.nativeAcknowledged or not axis.desiredLateral or not axis.lateralActive or
not sm.seen['aolAxisState'] or not sm.valid['aolAxisState'] or not sm.alive['aolAxisState'] or
not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
not self.pending_event and now_ns - stamp > AXIS_MAX_AGE_NS or
stamp - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
axis.validUntilMonoTime != axis.sourceCarStateMonoTime + AXIS_MAX_AGE_NS or
not stamp <= axis.validUntilMonoTime < stamp + AXIS_MAX_AGE_NS or
native.observedMonoTime < self.healthy_native_ns or
(native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory)) != self.healthy_inventory):
cp_identity(cp) != self.cp or
str(axis.sessionId) != self.session or
str(axis.faultSessionId) != self.session or str(axis.faultReason) != 'none' or
not axis.qualified or not axis.nativeAcknowledged or not axis.desiredLateral or not axis.lateralActive or
not sm.seen['aolAxisState'] or not sm.valid['aolAxisState'] or not sm.alive['aolAxisState'] or
not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
not self.pending_event and now_ns - stamp > AXIS_MAX_AGE_NS or
stamp - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS or
axis.validUntilMonoTime != axis.sourceCarStateMonoTime + AXIS_MAX_AGE_NS or
not stamp <= axis.validUntilMonoTime < stamp + AXIS_MAX_AGE_NS or
native.observedMonoTime < self.healthy_native_ns or
(native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory)) != self.healthy_inventory):
return False
event_ns = int(sm.logMonoTime['onroadEvents'])
faults = [event for event in sm['onroadEvents'] if event.immediateDisable or event.softDisable]
@@ -1258,17 +1258,17 @@ class TransportPauseFeedback:
combined_stop = bool(ccnc_stop and axis.desiredLongitudinal and axis.longitudinalActive)
subaru_stop = cs.standstill and subaru_restriction(cp, cs)
if (not (qualified(cp, marked_only=True) or mazda_qualified(cp, marked_only=True) or subaru_stop or ccnc_stop) or
cs.gearShifter != car.CarState.GearShifter.drive or
not (temporary_restriction(cp, cs) or mazda_restriction(cp, cs) or subaru_stop) 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 ccnc_stop) 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 subaru_stop and not axis.lateralActive) or not axis.nativeAcknowledged or
(axis.desiredLongitudinal or axis.longitudinalActive) and not combined_stop or
not 0 < axis.sourceCarStateMonoTime <= stamp <= now_ns or
now_ns - axis.sourceCarStateMonoTime > AXIS_MAX_AGE_NS):
cs.gearShifter != car.CarState.GearShifter.drive or
not (temporary_restriction(cp, cs) or mazda_restriction(cp, cs) or subaru_stop) 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 ccnc_stop) 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 subaru_stop and not axis.lateralActive) or not axis.nativeAcknowledged or
(axis.desiredLongitudinal or axis.longitudinalActive) and not combined_stop 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
@@ -1295,30 +1295,30 @@ class TransportPauseFeedback:
from opendbc.car import structs
previous = self.signature
if (not latched or self.failed or self.paused or self.pending_event or self.retention_started_ns or
self.source_retention_started_ns or not self.ever_lateral_active or previous is None or
not self.session or cp_identity(cp) != self.cp or previous[0] != self.session or
previous[7] != self.session or previous[5] != 'none' or any(previous[9:13]) or not previous[13] or
not producer or previous[14] != producer or
not 0 < previous[8] <= previous[2] <= self.healthy_feedback_ns <= now_ns or
now_ns - previous[8] > SAFETY_MAX_AGE_NS or
cs.gearShifter not in (structs.CarState.GearShifter.reverse, structs.CarState.GearShifter.neutral) or
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or
cs.accFaulted or cs.vehicleSensorsInvalid or cs.doorOpen or cs.seatbeltUnlatched or
cs.steeringDisengage or not cs.cruiseState.available or cs.buttonEvents or
not sm.seen['aolAxisState'] or not sm.alive['aolAxisState'] or sm.valid['aolAxisState'] or
axis.qualified or axis.sourceCarStateMonoTime != 0 or str(axis.mode) != 'off' or
axis.desiredLateral or axis.desiredLongitudinal or axis.lateralActive or axis.longitudinalActive or
axis.nativeAcknowledged or str(axis.faultReason) != 'none' or
str(axis.sessionId) != self.session or str(axis.faultSessionId) != self.session or
str(axis.intentProducerSessionId) != producer or int(axis.intentIntegrityEpoch) != previous[15] or
axis.sequence <= self.sequence or not previous[2] < stamp == axis.observedMonoTime <= now_ns <= axis.validUntilMonoTime or
axis.validUntilMonoTime - stamp != AXIS_MAX_AGE_NS):
self.source_retention_started_ns or not self.ever_lateral_active or previous is None or
not self.session or cp_identity(cp) != self.cp or previous[0] != self.session or
previous[7] != self.session or previous[5] != 'none' or any(previous[9:13]) or not previous[13] or
not producer or previous[14] != producer or
not 0 < previous[8] <= previous[2] <= self.healthy_feedback_ns <= now_ns or
now_ns - previous[8] > SAFETY_MAX_AGE_NS or
cs.gearShifter not in (structs.CarState.GearShifter.reverse, structs.CarState.GearShifter.neutral) or
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or
cs.accFaulted or cs.vehicleSensorsInvalid or cs.doorOpen or cs.seatbeltUnlatched or
cs.steeringDisengage or not cs.cruiseState.available or cs.buttonEvents or
not sm.seen['aolAxisState'] or not sm.alive['aolAxisState'] or sm.valid['aolAxisState'] or
axis.qualified or axis.sourceCarStateMonoTime != 0 or str(axis.mode) != 'off' or
axis.desiredLateral or axis.desiredLongitudinal or axis.lateralActive or axis.longitudinalActive or
axis.nativeAcknowledged or str(axis.faultReason) != 'none' or
str(axis.sessionId) != self.session or str(axis.faultSessionId) != self.session or
str(axis.intentProducerSessionId) != producer or int(axis.intentIntegrityEpoch) != previous[15] or
axis.sequence <= self.sequence or not previous[2] < stamp == axis.observedMonoTime <= now_ns <= axis.validUntilMonoTime or
axis.validUntilMonoTime - stamp != AXIS_MAX_AGE_NS):
return False
event_ns = int(sm.logMonoTime['onroadEvents'])
allowed = (log.OnroadEvent.EventName.wrongGear, log.OnroadEvent.EventName.reverseGear,
log.OnroadEvent.EventName.preEnableStandstill, log.OnroadEvent.EventName.pedalPressed,
log.OnroadEvent.EventName.gasPressedOverride, log.OnroadEvent.EventName.steerOverride,
log.OnroadEvent.EventName.pcmDisable)
log.OnroadEvent.EventName.preEnableStandstill, log.OnroadEvent.EventName.pedalPressed,
log.OnroadEvent.EventName.gasPressedOverride, log.OnroadEvent.EventName.steerOverride,
log.OnroadEvent.EventName.pcmDisable)
if (not sm.seen['onroadEvents'] or not sm.valid['onroadEvents'] or not sm.alive['onroadEvents'] or
not 0 < event_ns == axis.faultEventMonoTime <= stamp or now_ns - event_ns > 1_500_000_000 or
any(event.name not in allowed or event.immediateDisable or
@@ -1338,25 +1338,25 @@ class TransportPauseFeedback:
from opendbc.car import structs
previous = self.signature
if (not latched or self.failed or self.paused or self.pending_event or self.retention_started_ns or
self.source_retention_started_ns or not self.ever_lateral_active or previous is None or
not self.session or cp_identity(cp) != self.cp or previous[0] != self.session or
previous[7] != self.session or previous[5] != 'none' or not previous[11] or not previous[13] or
not producer or previous[14] != producer or
not 0 < previous[8] <= previous[2] <= self.healthy_feedback_ns <= now_ns or
now_ns - previous[8] > SAFETY_MAX_AGE_NS or
cs.gearShifter not in (structs.CarState.GearShifter.drive, structs.CarState.GearShifter.low,
structs.CarState.GearShifter.sport, structs.CarState.GearShifter.brake) or
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or
cs.accFaulted or cs.vehicleSensorsInvalid or cs.doorOpen or cs.seatbeltUnlatched or
cs.steeringDisengage or not cs.cruiseState.available or cs.buttonEvents or
not sm.seen['aolAxisState'] or not sm.alive['aolAxisState'] or not sm.valid['aolAxisState'] or
not axis.qualified or axis.sourceCarStateMonoTime != 0 or str(axis.mode) != 'off' or
axis.desiredLateral or axis.desiredLongitudinal or axis.lateralActive or axis.longitudinalActive or
axis.nativeAcknowledged or str(axis.faultReason) != 'availabilityPause' or
str(axis.sessionId) != self.session or str(axis.faultSessionId) != self.session or
str(axis.intentProducerSessionId) != producer or int(axis.intentIntegrityEpoch) != previous[15] or
axis.sequence <= self.sequence or not previous[2] < stamp == axis.observedMonoTime <= now_ns <= axis.validUntilMonoTime or
axis.validUntilMonoTime - stamp != AXIS_MAX_AGE_NS or not 0 < axis.faultEventMonoTime <= stamp):
self.source_retention_started_ns or not self.ever_lateral_active or previous is None or
not self.session or cp_identity(cp) != self.cp or previous[0] != self.session or
previous[7] != self.session or previous[5] != 'none' or not previous[11] or not previous[13] or
not producer or previous[14] != producer or
not 0 < previous[8] <= previous[2] <= self.healthy_feedback_ns <= now_ns or
now_ns - previous[8] > SAFETY_MAX_AGE_NS or
cs.gearShifter not in (structs.CarState.GearShifter.drive, structs.CarState.GearShifter.low,
structs.CarState.GearShifter.sport, structs.CarState.GearShifter.brake) or
not cs.canValid or cs.canTimeout or cs.steerFaultTemporary or cs.steerFaultPermanent or
cs.accFaulted or cs.vehicleSensorsInvalid or cs.doorOpen or cs.seatbeltUnlatched or
cs.steeringDisengage or not cs.cruiseState.available or cs.buttonEvents or
not sm.seen['aolAxisState'] or not sm.alive['aolAxisState'] or not sm.valid['aolAxisState'] or
not axis.qualified or axis.sourceCarStateMonoTime != 0 or str(axis.mode) != 'off' or
axis.desiredLateral or axis.desiredLongitudinal or axis.lateralActive or axis.longitudinalActive or
axis.nativeAcknowledged or str(axis.faultReason) != 'availabilityPause' or
str(axis.sessionId) != self.session or str(axis.faultSessionId) != self.session or
str(axis.intentProducerSessionId) != producer or int(axis.intentIntegrityEpoch) != previous[15] or
axis.sequence <= self.sequence or not previous[2] < stamp == axis.observedMonoTime <= now_ns <= axis.validUntilMonoTime or
axis.validUntilMonoTime - stamp != AXIS_MAX_AGE_NS or not 0 < axis.faultEventMonoTime <= stamp):
return False
event_ns = int(sm.logMonoTime['onroadEvents'])
events = sm['onroadEvents']
@@ -1395,17 +1395,17 @@ class TransportPauseFeedback:
stamp = int(sm.logMonoTime['aolAxisState'])
reason = str(axis.faultReason)
signature = (str(axis.sessionId), int(axis.sequence), stamp, int(axis.observedMonoTime),
int(axis.validUntilMonoTime), reason, int(axis.faultEventMonoTime), str(axis.faultSessionId),
int(axis.sourceCarStateMonoTime), bool(axis.lateralActive), bool(axis.longitudinalActive),
bool(axis.desiredLateral), bool(axis.desiredLongitudinal), bool(axis.nativeAcknowledged),
str(getattr(axis, 'intentProducerSessionId', '')), int(getattr(axis, 'intentIntegrityEpoch', 0)))
int(axis.validUntilMonoTime), reason, int(axis.faultEventMonoTime), str(axis.faultSessionId),
int(axis.sourceCarStateMonoTime), bool(axis.lateralActive), bool(axis.longitudinalActive),
bool(axis.desiredLateral), bool(axis.desiredLongitudinal), bool(axis.nativeAcknowledged),
str(getattr(axis, 'intentProducerSessionId', '')), int(getattr(axis, 'intentIntegrityEpoch', 0)))
current = bool(sm.seen['aolAxisState'] and sm.valid['aolAxisState'] and sm.alive['aolAxisState'] and
axis.qualified and axis.sessionId and axis.sequence > 0 and stamp == axis.observedMonoTime and
0 < stamp <= now_ns <= axis.validUntilMonoTime and now_ns - stamp <= AXIS_MAX_AGE_NS and
axis.validUntilMonoTime - stamp <= AXIS_MAX_AGE_NS and axis.faultSessionId == axis.sessionId and
cp_identity(cp) == self.cp)
axis.qualified and axis.sessionId and axis.sequence > 0 and stamp == axis.observedMonoTime and
0 < stamp <= now_ns <= axis.validUntilMonoTime and now_ns - stamp <= AXIS_MAX_AGE_NS and
axis.validUntilMonoTime - stamp <= AXIS_MAX_AGE_NS and axis.faultSessionId == axis.sessionId and
cp_identity(cp) == self.cp)
continuity = bool(not self.session or axis.sessionId == self.session and
(axis.sequence > self.sequence or axis.sequence == self.sequence and signature == self.signature))
(axis.sequence > self.sequence or axis.sequence == self.sequence and signature == self.signature))
if not current and continuity and self._inactive_gear_source_gap(sm, cp, cs, axis, stamp, now_ns, latched=latched,
producer=intent_producer_session):
return False
@@ -1678,8 +1678,8 @@ class TransportPauseFeedback:
axis.faultEventMonoTime == int(sm.logMonoTime['onroadEvents']) and
main_off_idle_native_baseline(cp, cs, axis, native, sm['onroadEvents'], producer=intent_producer_session))
baseline = bool(reason == 'none' and (axis.nativeAcknowledged or inactive or gm_main_off) and
0 < axis.sourceCarStateMonoTime <= stamp and
stamp - axis.sourceCarStateMonoTime <= AXIS_MAX_AGE_NS)
0 < axis.sourceCarStateMonoTime <= stamp and
stamp - axis.sourceCarStateMonoTime <= AXIS_MAX_AGE_NS)
if native is None and latched and current and continuity and baseline:
native = _newer_combined_intent_native(sm, cp, cs, axis, now_ns)
lateral_native = (self._native_long_withdrawal_lateral_current(sm, cp, cs, axis, now_ns,
@@ -1705,8 +1705,8 @@ class TransportPauseFeedback:
return False
transport_healthy = native is not None
cancel_native = (self._longitudinal_cancel_current(sm, cp, cs, axis, now_ns, latched=latched,
cancel_source_ns=cancel_source_ns)
if current and continuity and baseline and not transport_healthy else None)
cancel_source_ns=cancel_source_ns)
if current and continuity and baseline and not transport_healthy else None)
if self._event_first_retention(sm, cp, axis, stamp, now_ns, latched=latched,
signature=signature, native=native):
return False
@@ -1791,10 +1791,10 @@ class TransportPauseFeedback:
self.healthy_feedback_ns = now_ns
self.healthy_native_ns = int(native.observedMonoTime) if native is not None else 0
self.healthy_inventory = ((native.pandaSerial, tuple(p.hardwareSerial for p in native.pandaInventory))
if native is not None else None)
if native is not None else None)
self.session = str(axis.sessionId)
lateral_baseline = bool(axis.desiredLateral and axis.lateralActive and native is not None and
native.pandaInventory and native.requestedLateral and native.lateralAllowed)
native.pandaInventory and native.requestedLateral and native.lateralAllowed)
self.lateral_baseline_ns = now_ns if lateral_baseline else 0
self.lateral_source_ns = int(axis.sourceCarStateMonoTime) if lateral_baseline else 0
self.availability_producer = (str(getattr(axis, 'intentProducerSessionId', '')) if lateral_baseline and
@@ -1818,15 +1818,15 @@ class TransportPauseFeedback:
self.paused = False
else:
provenance = bool(known and not axis.lateralActive and not axis.longitudinalActive and
not axis.desiredLongitudinal and 0 < axis.faultEventMonoTime <= stamp)
not axis.desiredLongitudinal and 0 < axis.faultEventMonoTime <= stamp)
pending = bool(provenance and axis.faultEventMonoTime > sm.logMonoTime['onroadEvents'])
if pending and not self.pending_event:
self.pending_event = int(axis.faultEventMonoTime)
self.qualified = bool(provenance and
axis.faultEventMonoTime == sm.logMonoTime['onroadEvents'] and
sm.valid['onroadEvents'] and axis.faultEventMonoTime <= now_ns)
axis.faultEventMonoTime == sm.logMonoTime['onroadEvents'] and
sm.valid['onroadEvents'] and axis.faultEventMonoTime <= now_ns)
if self.pending_event and (axis.faultEventMonoTime != self.pending_event or
now_ns - self.pending_event > AXIS_MAX_AGE_NS):
now_ns - self.pending_event > AXIS_MAX_AGE_NS):
self.pending_event = 0
self.paused = False
self.failed = True
openpilot/starpilot/galaxy/tests/test_settings.py
+40 -40
View File
@@ -42,10 +42,10 @@ class SettingsGatewayTest(unittest.TestCase):
self.params = Params(temporary.name)
native = ToyotaInterface.get_non_essential_params(ToyotaCar.TOYOTA_COROLLA_TSS2)
cp = SimpleNamespace(carFingerprint="TOYOTA_COROLLA_TSS2", openpilotLongitudinalControl=True,
pcmCruise=False, notCar=False, dashcamOnly=False, passive=False,
brand="toyota", steerControlType=native.steerControlType, carVin="",
lateralTuning=native.lateralTuning, minSteerSpeed=native.minSteerSpeed, safetyConfigs=native.safetyConfigs,
transmissionType=car.CarParams.TransmissionType.automatic)
pcmCruise=False, notCar=False, dashcamOnly=False, passive=False,
brand="toyota", steerControlType=native.steerControlType, carVin="",
lateralTuning=native.lateralTuning, minSteerSpeed=native.minSteerSpeed, safetyConfigs=native.safetyConfigs,
transmissionType=car.CarParams.TransmissionType.automatic)
self.context = MutableContext(cp)
self.gateway = SettingsGateway(self.params, self.context, clock=lambda: 100.0)
self.session, self.generation = "local-session", b"generation"
@@ -460,7 +460,7 @@ class SettingsGatewayTest(unittest.TestCase):
self.params.put('SLCPriority1', 'Dashboard', block=True)
self.params.put('SLCPriority2', 'Dashboard', block=True)
for replay, vision, available in (("0", "0", False), ("1", "0", False),
("0", "1", False), ("1", "1", True)):
("0", "1", False), ("1", "1", True)):
with self.subTest(replay=replay, vision=vision), mock.patch.dict(
os.environ, {"SLC_REPLAY_RUNTIME": replay, "SLC_VISION_DEVELOPMENT": vision}):
self.assertEqual(diagnostic_choice_enabled(os.environ), available)
@@ -1133,10 +1133,10 @@ class SettingsGatewayTest(unittest.TestCase):
def test_live_lane_keys_write_through_existing_owner(self):
_, root = self._live_lane_context()
for key, value, expected in (("LaneCentering", "On", b"1"),
("LaneCenteringPauseOnSignal", "Off", b"0"),
("LaneCenterOffset", 0.15, b"0.15"),
("LaneCenteringE2EAuthority", 75, b"0.75"),
("LaneCenteringStrength", 105, b"1.575")):
("LaneCenteringPauseOnSignal", "Off", b"0"),
("LaneCenterOffset", 0.15, b"0.15"),
("LaneCenteringE2EAuthority", 75, b"0.75"),
("LaneCenteringStrength", 105, b"1.575")):
with self.subTest(key=key):
page = self.page("lane")
labels = {"LaneCentering": "Enable Lane Centering", "LaneCenteringPauseOnSignal": "Pause on signal",
@@ -1164,7 +1164,7 @@ class SettingsGatewayTest(unittest.TestCase):
self.context.value = AuthorityContext(False, bad, bad.as_reader().as_builder().to_bytes())
self.assertFalse(any(row["available"] for row in self.page("lane")["rows"]))
self.context.value = AuthorityContext(False, SimpleNamespace(brand="unknown", carFingerprint="OTHER", notCar=False,
dashcamOnly=False, passive=False), b"other")
dashcamOnly=False, passive=False), b"other")
page = self.page("lane")
rows = page["rows"]
self.assertFalse(next(row for row in rows if row["label"] == "Lane Centering Strength")["available"])
@@ -1431,8 +1431,8 @@ class LiveContextTest(unittest.TestCase):
self.wait_now += timeout / 1000
messages.update = mock.Mock(side_effect=wait_update)
source = LiveContextSource(self.params, messages, mono_clock=lambda: self.mono,
boot_clock=lambda: self.boot, evidence_wait_ms=20,
wait_clock=lambda: self.wait_now)
boot_clock=lambda: self.boot, evidence_wait_ms=20,
wait_clock=lambda: self.wait_now)
self.assertFalse(source.parked())
self.assertEqual(messages.calls[2], 0)
self.assertTrue(any(timeout > 0 for timeout in messages.calls[3:]))
@@ -1513,10 +1513,10 @@ class SettingsHttpTest(unittest.TestCase):
self.params = Params(temporary.name)
native = ToyotaInterface.get_non_essential_params(ToyotaCar.TOYOTA_COROLLA_TSS2)
cp = SimpleNamespace(carFingerprint="TOYOTA_COROLLA_TSS2", openpilotLongitudinalControl=True,
pcmCruise=False, notCar=False, dashcamOnly=False, passive=False,
brand="toyota", steerControlType=native.steerControlType, carVin="",
lateralTuning=native.lateralTuning, minSteerSpeed=native.minSteerSpeed, safetyConfigs=native.safetyConfigs,
transmissionType=car.CarParams.TransmissionType.automatic)
pcmCruise=False, notCar=False, dashcamOnly=False, passive=False,
brand="toyota", steerControlType=native.steerControlType, carVin="",
lateralTuning=native.lateralTuning, minSteerSpeed=native.minSteerSpeed, safetyConfigs=native.safetyConfigs,
transmissionType=car.CarParams.TransmissionType.automatic)
self.context = MutableContext(cp)
self.gateway = SettingsGateway(self.params, self.context)
self.access = GalaxyAccessOwner(Path(temporary.name) / "access")
@@ -1559,7 +1559,7 @@ class SettingsHttpTest(unittest.TestCase):
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Force Stop")
self.assertTrue(page["rows"][index]["available"])
status, intent, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": index, "value": "Off"})
payload={"view": page["view"], "row": index, "value": "Off"})
self.assertEqual(status, 200)
self.assertEqual(self.request("/api/settings/confirm", cookie=cookie,
payload={"intent": intent["intent"], "confirmed": True})[:2], (200, {"saved": True}))
@@ -1584,7 +1584,7 @@ class SettingsHttpTest(unittest.TestCase):
index = next(i for i, row in enumerate(page['rows']) if row['label'] == 'Distance short press')
self.assertTrue(page['rows'][index]['available'])
status, intent, _ = self.request('/api/settings/preview', cookie=cookie,
payload={'view': page['view'], 'row': index, 'value': 'Traffic mode'})
payload={'view': page['view'], 'row': index, 'value': 'Traffic mode'})
self.assertEqual(status, 200)
return intent['intent']
intent = preview()
@@ -1618,7 +1618,7 @@ class SettingsHttpTest(unittest.TestCase):
self.assertEqual(status, 200)
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Selected Deceleration Profile")
status, result, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": index, "value": "Comfort"})
payload={"view": page["view"], "row": index, "value": "Comfort"})
self.assertEqual(status, 200)
return result["intent"]
@@ -1683,7 +1683,7 @@ class SettingsHttpTest(unittest.TestCase):
self.assertEqual(status, 200)
warning = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Immediate Warning")
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": warning, "direction": 1})
payload={"view": page["view"], "row": warning, "direction": 1})
self.assertEqual(status, 200)
self.assertEqual(preview["proposed"], "25%")
self.assertIsNone(self.params.get("WarningImmediateVolume"))
@@ -1696,7 +1696,7 @@ class SettingsHttpTest(unittest.TestCase):
self.assertEqual(status, 200)
master = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Custom Display Settings")
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": master, "direction": 1})
payload={"view": page["view"], "row": master, "direction": 1})
self.assertEqual(status, 200)
self.context.value = AuthorityContext(False, None, None)
self.assertEqual(self.request("/api/settings/confirm", cookie=cookie,
@@ -1715,7 +1715,7 @@ class SettingsHttpTest(unittest.TestCase):
blindspot = next(i for i, row in enumerate(page["rows"]) if row["label"] == "C4 red blind-spot border")
self.assertEqual([page["rows"][index]["value"] for index in (signal, blindspot)], ["Off", "On"])
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": signal, "direction": 1})
payload={"view": page["view"], "row": signal, "direction": 1})
self.assertEqual(status, 200)
self.assertIsNone(read_visibility(self.params, "SignalMetrics").raw)
self.assertEqual(self.request("/api/settings/confirm", cookie=cookie,
@@ -1725,7 +1725,7 @@ class SettingsHttpTest(unittest.TestCase):
page = self.request("/api/settings/pages/appearance", cookie=cookie)[1]
blindspot = next(i for i, row in enumerate(page["rows"]) if row["label"] == "C4 red blind-spot border")
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": blindspot, "direction": -1})
payload={"view": page["view"], "row": blindspot, "direction": -1})
self.assertEqual(status, 200)
self.context.value = AuthorityContext(False, None, None)
self.assertEqual(self.request("/api/settings/confirm", cookie=cookie,
@@ -1744,11 +1744,11 @@ class SettingsHttpTest(unittest.TestCase):
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Restore default camera crop")
self.assertTrue(page["rows"][index]["confirm"])
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": index, "direction": 0})
payload={"view": page["view"], "row": index, "direction": 0})
self.assertEqual(status, 200)
self.assertIsNone(self.params.get(MASK))
status, result, _ = self.request("/api/settings/confirm", cookie=cookie,
payload={"intent": preview["intent"], "confirmed": True})
payload={"intent": preview["intent"], "confirmed": True})
self.assertEqual((status, result), (200, {"saved": True}))
saved = read_pip(self.params)
self.assertEqual((saved.mask.width, saved.mask.height), (1928, 1208))
@@ -1806,16 +1806,16 @@ class SettingsHttpTest(unittest.TestCase):
self.assertEqual(status, 200)
self.assertEqual(page["rows"][0]["value"], "Off")
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": 0, "direction": 1})
payload={"view": page["view"], "row": 0, "direction": 1})
self.assertEqual(status, 200)
self.assertFalse(Path(self.params.get_param_path(KEY)).exists())
status, result, _ = self.request("/api/settings/confirm", cookie=cookie,
payload={"intent": preview["intent"], "confirmed": True})
payload={"intent": preview["intent"], "confirmed": True})
self.assertEqual((status, result), (200, {"saved": True}))
self.assertTrue(decode(Path(self.params.get_param_path(KEY)).read_bytes()).enabled)
page = self.request("/api/settings/pages/sentry", cookie=cookie)[1]
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": 1, "direction": 1})
payload={"view": page["view"], "row": 1, "direction": 1})
self.assertEqual(status, 200)
self.assertEqual(self.request("/api/auth/logout", cookie=cookie, payload={})[0], 200)
self.assertEqual(self.request("/api/settings/confirm", cookie=cookie,
@@ -1833,7 +1833,7 @@ class SettingsHttpTest(unittest.TestCase):
self.assertEqual(status, 200)
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Show on turn signal")
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": index, "direction": 1})
payload={"view": page["view"], "row": index, "direction": 1})
self.assertEqual(status, 200)
if failure == "native_change":
Path(self.params.get_param_path(MASK)).write_bytes(encode_mask(starting_mask(1344, 760)))
@@ -1853,11 +1853,11 @@ class SettingsHttpTest(unittest.TestCase):
self.assertEqual(status, 200)
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Curve Speed Controller")
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": index, "direction": 1})
payload={"view": page["view"], "row": index, "direction": 1})
self.assertEqual(status, 200)
self.assertFalse(self.params.get_bool("CurveSpeedController"))
status, result, _ = self.request("/api/settings/confirm", cookie=cookie,
payload={"intent": preview["intent"], "confirmed": True})
payload={"intent": preview["intent"], "confirmed": True})
self.assertEqual((status, result), (200, {"saved": True}))
self.assertTrue(self.params.get_bool("CurveSpeedController"))
@@ -1869,12 +1869,12 @@ class SettingsHttpTest(unittest.TestCase):
self.assertEqual(status, 200)
self.assertEqual(page["rows"][0]["value"], "Conditional Experimental")
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": 0, "direction": 1})
payload={"view": page["view"], "row": 0, "direction": 1})
self.assertEqual(status, 200)
self.assertIsNone(self.params.get("ConditionalModeConfig"))
self.context.value = AuthorityContext(False, self.context.value.cp, self.context.value.cp_raw)
status, result, _ = self.request("/api/settings/confirm", cookie=cookie,
payload={"intent": preview["intent"], "confirmed": True})
payload={"intent": preview["intent"], "confirmed": True})
self.assertEqual((status, result), (200, {"saved": True}))
saved = Path(self.params.get_param_path("ConditionalModeConfig")).read_bytes()
self.assertEqual(decode_preferences(saved).mode.value, "conditional_chill")
@@ -1889,10 +1889,10 @@ class SettingsHttpTest(unittest.TestCase):
self.assertEqual(status, 200)
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Remember manual choice")
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": index, "direction": 1})
payload={"view": page["view"], "row": index, "direction": 1})
self.assertEqual(status, 200)
status, result, _ = self.request("/api/settings/confirm", cookie=cookie,
payload={"intent": preview["intent"], "confirmed": True})
payload={"intent": preview["intent"], "confirmed": True})
self.assertEqual((status, result), (200, {"saved": True}))
self.assertEqual(decode(manual.read_bytes()), SavedCodes(cem=0, ccm=1))
self.assertTrue(decode_preferences(Path(self.params.get_param_path("ConditionalModeConfig")).read_bytes()).cem.persist_manual)
@@ -1907,11 +1907,11 @@ class SettingsHttpTest(unittest.TestCase):
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Reset Saved Curve Learning")
self.assertTrue(page["rows"][index]["confirm"])
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": index, "direction": 0})
payload={"view": page["view"], "row": index, "direction": 0})
self.assertEqual(status, 200)
self.assertEqual(Path(self.params.get_param_path("CurveComfortData")).read_bytes(), b"{")
status, result, _ = self.request("/api/settings/confirm", cookie=cookie,
payload={"intent": preview["intent"], "confirmed": True})
payload={"intent": preview["intent"], "confirmed": True})
self.assertEqual((status, result), (200, {"saved": True}))
self.assertEqual(Path(self.params.get_param_path("CurveComfortData")).read_bytes(), b'{"version":1,"buckets":{}}')
self.assertEqual(legacy.read_bytes(), b'{}')
@@ -1935,7 +1935,7 @@ class SettingsHttpTest(unittest.TestCase):
self.assertEqual(self.request("/api/settings/pages/aggressive%2Ffollowing", cookie=cookie)[0], 200)
index = next(i for i, row in enumerate(page["rows"]) if row["label"] == "Enable Lane Centering")
status, preview, _ = self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": index, "direction": 1})
payload={"view": page["view"], "row": index, "direction": 1})
self.assertEqual(status, 200)
self.assertFalse(self.params.get_bool("LaneCentering"))
status, saved, _ = self.request("/api/settings/confirm", cookie=cookie,
@@ -1975,8 +1975,8 @@ class SettingsHttpTest(unittest.TestCase):
payload={"intent": "invented", "confirmed": True})[0], 409)
self.assertFalse(self.params.get_bool("LaneCentering"))
self.assertEqual(self.request("/api/settings/preview", cookie=cookie,
payload={"view": page["view"], "row": index, "direction": 1},
origin="http://malicious.local")[0], 403)
payload={"view": page["view"], "row": index, "direction": 1},
origin="http://malicious.local")[0], 403)
self.assertFalse(self.params.get_bool("LaneCentering"))
def test_confirm_finishes_before_acknowledged_logout(self):
prebuilt.json
+1 -1
View File
@@ -1 +1 @@
{"protected_inventory_sha256": "730435310bd927cef4fd4a56d0ad030df3e4f14326f8c3397d614b4b703a2791", "schema_version": 1}
{"protected_inventory_sha256": "f369a36361ba34767db0f59f0f36043be8b2de503bcbe15b287d0eca0074af99", "schema_version": 1}