mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-29 09:33:48 +08:00
Restore recorded Sonata Hybrid diagnostic admission
This commit is contained in:
firestar5683
@@ -1,4 +1,4 @@
|
||||
"""Compatibility policy for qualified Bosch Honda AOL and two recorded GM configurations; native authority stays separate.
|
||||
"""Compatibility policy for recorded diagnostic configurations; native authority stays separate.
|
||||
|
||||
Register divergence remains a protected-register diagnostic, not a benign-fault
|
||||
claim. Raw fault reports are retained. Every unlisted configuration/bit blocks.
|
||||
@@ -16,6 +16,16 @@ def _diagnostic_mask(cp):
|
||||
# requires a marked, supported, single-panda Bosch AOL configuration.
|
||||
# Other interrupt rates and mixed faults remain blocking.
|
||||
return (1 << 3) if int(cp.flags) & int(HondaFlags.BOSCH) and qualified(cp, marked_only=True) else 0
|
||||
if cp.brand == 'hyundai':
|
||||
from opendbc.car.hyundai.values import CAR, HyundaiFlags
|
||||
if cp.carFingerprint != CAR.HYUNDAI_SONATA_HYBRID:
|
||||
return 0
|
||||
from opendbc.car.hyundai.classic_scc_aol import qualified
|
||||
recorded_flags = (HyundaiFlags.HYBRID | HyundaiFlags.SEND_LFA | HyundaiFlags.USE_FCA |
|
||||
HyundaiFlags.MANDO_RADAR | HyundaiFlags.CHECKSUM_CRC8 | HyundaiFlags.HAS_LDA_BUTTON)
|
||||
return (1 << 3) if (int(cp.flags) == int(recorded_flags) and
|
||||
len(cp.safetyConfigs) == 1 and int(cp.safetyConfigs[0].safetyParam) == 0xC02 and cp.alternativeExperience == 32 and
|
||||
qualified(cp, marked_only=True)) else 0
|
||||
from opendbc.car.gm.aol import qualified_gm
|
||||
from opendbc.car.gm.values import CAR, is_volt_sdgm_profile, is_bolt_present_no_acc_pedal_profile
|
||||
if cp.brand != 'gm' or len(cp.safetyConfigs) != 1:
|
||||
|
||||
@@ -0,0 +1,384 @@
|
||||
import dataclasses
|
||||
import json
|
||||
import pytest
|
||||
import importlib
|
||||
import os
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
from opendbc.can import CANPacker
|
||||
from opendbc.car import gen_empty_fingerprint, structs
|
||||
from opendbc.car.hyundai.interface import CarInterface
|
||||
from opendbc.car.hyundai.radar_interface import RadarInterface
|
||||
from opendbc.car.hyundai.values import CAR
|
||||
from opendbc.car.hyundai.classic_scc_aol import qualified as qualified_classic
|
||||
from opendbc.safety.tests.libsafety import libsafety_py
|
||||
from opendbc.safety.tests.test_hyundai import checksum
|
||||
from openpilot.cereal import messaging
|
||||
from openpilot.common.params import Params
|
||||
from openpilot.common.prefix import OpenpilotPrefix
|
||||
from openpilot.selfdrive.car import card as card_module
|
||||
from openpilot.starpilot.aol.intent import AOL_TOGGLE, read_settings
|
||||
from openpilot.starpilot.aol.runtime import current_native
|
||||
|
||||
from openpilot.starpilot.aol.tests import test_ioniq_availability as caller_fixture
|
||||
from openpilot.starpilot.aol.wire import PandaSlot, SafetyState, encode_safety
|
||||
from openpilot.starpilot.car.hyundai.classic_scc_intent import ClassicSccCardIntent
|
||||
from openpilot.cereal import log
|
||||
|
||||
disable_ecu_module = importlib.import_module('opendbc.car.disable_ecu')
|
||||
|
||||
|
||||
def classic_frames(ci, tick, *, main=0, lkas=0, button=0, engaged=False):
|
||||
|
||||
assert ci.CP.carFingerprint == CAR.HYUNDAI_SONATA_HYBRID
|
||||
packer = CANPacker('hyundai_can_generated')
|
||||
|
||||
|
||||
hz = {'E_EMS11':100, 'ELECT_GEAR':20, 'SCC11':50, 'SCC12':50, 'WHL_SPD11':100, 'TCS13':100, 'MDPS12':50, 'CLU11':50,
|
||||
'BCM_PO_11':50, 'LKAS11':100}
|
||||
scheduled = {(packer.dbc.name_to_msg[name].address, 0): name for name in hz if name != 'LKAS11'}
|
||||
for parser in ci.can_parsers.values():
|
||||
for address in parser.addresses:
|
||||
scheduled[(address, parser.bus)] = parser.dbc.addr_to_msg[address].name
|
||||
frames = []
|
||||
for (address,bus),name in sorted(scheduled.items()):
|
||||
parser = next((p for p in ci.can_parsers.values() if p.bus == bus), None)
|
||||
declared = parser.message_states[address].frequency if parser is not None and address in parser.addresses else 0
|
||||
frequency = hz.get(name, min(declared,100) if 0 < declared <= 100 else 100)
|
||||
period = max(1,round(100/frequency))
|
||||
if tick % period:
|
||||
continue
|
||||
counter = tick // period
|
||||
values = {}
|
||||
if name == 'CLU11':
|
||||
values.update(CF_Clu_AliveCnt1=counter%16, CF_Clu_CruiseSwState=button, CF_Clu_CruiseSwMain=main)
|
||||
elif name == 'E_EMS11':
|
||||
values.update(CR_Vcu_AccPedDep_Pos=0)
|
||||
elif name == 'WHL_SPD11':
|
||||
values.update(WHL_SPD_AliveCounter_LSB=counter%4, WHL_SPD_AliveCounter_MSB=(counter//4)%4,
|
||||
WHL_SPD_FL=40,WHL_SPD_FR=40,WHL_SPD_RL=40,WHL_SPD_RR=40)
|
||||
elif name == 'TCS13':
|
||||
values.update(AliveCounterTCS=counter%8, ACCEnable=0, ACC_REQ=int(engaged), DriverOverride=0)
|
||||
elif name == 'MDPS12':
|
||||
values.update(CR_Mdps_StrColTq=0, CR_Mdps_OutTq=0)
|
||||
elif name == 'BCM_PO_11':
|
||||
values['LDA_BTN']=lkas
|
||||
elif name == 'CGW1':
|
||||
values['CF_Gway_DrvSeatBeltSw']=1
|
||||
elif name == 'ELECT_GEAR':
|
||||
values['Elect_Gear_Shifter']=5
|
||||
elif name == 'SCC11':
|
||||
values['MainMode_ACC']=main
|
||||
elif name == 'SCC12':
|
||||
values.update(ACCMode=int(engaged), CR_VSM_Alive=counter%16)
|
||||
elif name == 'SAS11':
|
||||
values['SAS_Angle']=1.
|
||||
elif name == 'CLU15':
|
||||
values['CF_Clu_VehicleSpeed']=40
|
||||
frame = packer.make_can_msg(name,bus,values)
|
||||
if name in ('SCC12','WHL_SPD11','TCS13'):
|
||||
address, data, bus = checksum(frame)
|
||||
frame = (address, bytes(data), bus)
|
||||
frames.append(frame)
|
||||
assert not ci.CP.openpilotLongitudinalControl and ci.CP.pcmCruise
|
||||
assert all(address!=0x340 or bus==2 for address,_,bus in frames), 'Never echo outgoing LKAS on protected RX bus'
|
||||
return frames
|
||||
|
||||
|
||||
def _invalid_controls_emits_no_commands(received):
|
||||
if 'sendcan' not in received:
|
||||
return True
|
||||
envelope = received['sendcan']
|
||||
return envelope.which() == 'sendcan' and not envelope.valid and len(envelope.sendcan) == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize('raw_fault,denial', [
|
||||
(0, 'request'), (8, 'request'), (8, 'checksum'), (8, 'heartbeat'), (8, 'relay'),
|
||||
(8, 'frequency'), (8, 'stale-wire'), (8, 'telemetry-rate'),
|
||||
])
|
||||
def test_recorded_irq_full_stock_sonata_callers(raw_fault, denial, monkeypatch):
|
||||
shared_long = False
|
||||
native = libsafety_py.libsafety
|
||||
case = caller_fixture.IoniqAvailabilityTests(methodName='runTest')
|
||||
card = None
|
||||
with OpenpilotPrefix(), patch.dict(os.environ, {'SIMULATION': '0', 'AOL_REPLAY_RUNTIME': '1'}):
|
||||
saved = Params()
|
||||
saved.put_bool('OpenpilotEnabledToggle', True, block=True)
|
||||
saved.put_bool('AlwaysOnLateral', True, block=True)
|
||||
saved.put('LKASButtonControl', AOL_TOGGLE, block=True)
|
||||
saved.put('MainCruiseButtonControl', AOL_TOGGLE, block=True)
|
||||
now = [10_000_000_000]
|
||||
for key in ('CancelButtonControl', 'LongCancelButtonControl', 'VeryLongCancelButtonControl'):
|
||||
saved.put(key, 0, block=True)
|
||||
try:
|
||||
saved.put_bool('AlphaLongitudinalEnabled', False, block=True)
|
||||
saved.put_bool('IsReleaseBranch', False, block=True)
|
||||
fp = gen_empty_fingerprint()
|
||||
fp[0].update({0x391: 8, 0x38d: 8})
|
||||
fp[2][0x485] = 4
|
||||
fp[1][0x500] = 8
|
||||
cp = CarInterface.get_params(CAR.HYUNDAI_SONATA_HYBRID, fp, [], False, False, False)
|
||||
assert int(cp.flags) == 0x1011580
|
||||
assert qualified_classic(cp) and cp.pcmCruise and not cp.openpilotLongitudinalControl
|
||||
assert cp.lateralTuning.which() == 'torque' and not CarInterface.startup_required(cp)
|
||||
card = card_module.Car(CI=CarInterface(cp), RI=RadarInterface(cp))
|
||||
assert qualified_classic(card.CP, marked_only=True)
|
||||
assert card.CP.safetyConfigs[0].safetyParam == 0xC02 and card.CP.alternativeExperience == 32
|
||||
assert isinstance(card.aol_card_intent, ClassicSccCardIntent)
|
||||
assert not card.ci_initialized and not saved.get_bool('ControlsReady')
|
||||
card.slc_producer_session = 'authority-card'
|
||||
card.CI.update([])
|
||||
sd, controls, drive = case._authority_callers(False, cp=card.CP.as_reader().as_builder(), genuine_events=True)
|
||||
|
||||
sd.aol_settings = read_settings(saved)
|
||||
assert sd.aol_settings.lkas_action == AOL_TOGGLE and sd.aol_settings.main_action == AOL_TOGGLE
|
||||
assert sd.CP.to_dict() == card.CP.to_dict() == controls.CP.to_dict()
|
||||
assert len(card.CP.safetyConfigs) == 1 and bool(card.CP.openpilotLongitudinalControl) == shared_long
|
||||
config = card.CP.safetyConfigs[0]
|
||||
model, word = int(config.safetyModel.raw), int(config.safetyParam)
|
||||
native.set_alternative_experience(card.CP.alternativeExperience)
|
||||
assert native.set_safety_hooks(model, word) == 0
|
||||
native.init_tests()
|
||||
native.set_alternative_experience(card.CP.alternativeExperience)
|
||||
queued, received, previous_sd = [], {}, {}
|
||||
tick = 0
|
||||
native_counts = {'accepted': 0, 'blocked': 0, 'nonzero': 0, 'lateral_only_nonzero': 0}
|
||||
freeze_native = [None]
|
||||
last_native = [None]
|
||||
telemetry_rate_bad = [False]
|
||||
patch_scope = monkeypatch.context()
|
||||
local = patch_scope.__enter__()
|
||||
try:
|
||||
diagnostic_requests = []
|
||||
def diagnostic_query(sender, receiver, bus, addresses, requests, responses):
|
||||
assert bus == 0 and addresses == [(0x7d0, None)]
|
||||
assert len(requests) == 1
|
||||
request = requests[0]
|
||||
assert request in (b'\x10\x03', b'\x28\x83\x01')
|
||||
diagnostic_requests.append(request)
|
||||
def receive(timeout):
|
||||
sender([(0x7d0, bytes([len(request)]) + request + bytes(7-len(request)), bus)])
|
||||
return {(0x7d0, None): b'\x50\x03'} if request == b'\x10\x03' else {}
|
||||
return SimpleNamespace(get_data=receive)
|
||||
local.setattr(disable_ecu_module, 'IsoTpParallelQuery', diagnostic_query)
|
||||
local.setattr(card_module, 'REPLAY', False)
|
||||
local.setattr(card_module.time, 'monotonic_ns', lambda: now[0])
|
||||
local.setattr(card_module.time, 'clock_gettime_ns', lambda clock: now[0])
|
||||
local.setattr(card_module.messaging, 'drain_sock_raw', lambda *args, **kwargs: list(queued))
|
||||
local.setattr(card.sm, 'update', lambda *args: None)
|
||||
def panda_cadence(original):
|
||||
due = [0.0]
|
||||
def update(stamp, messages):
|
||||
delivered = []
|
||||
for message in messages:
|
||||
if message.which() == 'pandaStates':
|
||||
if stamp + 1e-9 < due[0]:
|
||||
continue
|
||||
due[0] = stamp + (0.0 if telemetry_rate_bad[0] else 0.1)
|
||||
delivered.append(message)
|
||||
return original(stamp, delivered)
|
||||
return update
|
||||
local.setattr(sd.sm, 'update_msgs', panda_cadence(sd.sm.update_msgs))
|
||||
local.setattr(card.sm, 'update_msgs', panda_cadence(card.sm.update_msgs))
|
||||
create_message = messaging.new_message
|
||||
def clocked_message(service=None, size=None, **kwargs):
|
||||
kwargs.setdefault('logMonoTime', now[0])
|
||||
event = create_message(service, size, **kwargs)
|
||||
if service == 'aolSafetyWire' and freeze_native[0] is not None:
|
||||
event.logMonoTime = freeze_native[0].observedMonoTime
|
||||
if service == 'pandaStates' and raw_fault:
|
||||
for panda in event.pandaStates:
|
||||
panda.faults = ['interruptRateCan2']
|
||||
return event
|
||||
local.setattr(messaging, 'new_message', clocked_message)
|
||||
original_encode = caller_fixture.encode_safety
|
||||
def recorded_fault_wire(state):
|
||||
state = dataclasses.replace(state, pandaInventory=tuple(
|
||||
dataclasses.replace(slot, faults=slot.faults | raw_fault) for slot in state.pandaInventory))
|
||||
last_native[0] = state
|
||||
return original_encode(freeze_native[0] or state)
|
||||
local.setattr(caller_fixture, 'encode_safety', recorded_fault_wire)
|
||||
send = card.pm.send
|
||||
def capture(service, event):
|
||||
raw = event if isinstance(event, bytes) else event.to_bytes()
|
||||
received[service] = messaging.log_from_bytes(raw).as_builder()
|
||||
return send(service, raw)
|
||||
local.setattr(card.pm, 'send', capture)
|
||||
|
||||
def step(*, main=1, lkas=0, button=0, engaged=False, withdraw_request=False, bad_checksum=False,
|
||||
heartbeat_override=None, inject_relay=False, drop_wheel=False):
|
||||
nonlocal tick
|
||||
tick += 1
|
||||
now[0] = source_ns = 10_000_000_000 + tick * 10_000_000
|
||||
prior_axis = previous_sd.get('aolAxisState')
|
||||
prior_state = previous_sd.get('selfdriveState')
|
||||
request = 0 if prior_axis is None else (
|
||||
int(prior_axis.aolAxisState.desiredLateral) | int(prior_axis.aolAxisState.desiredLongitudinal) << 1)
|
||||
if withdraw_request:
|
||||
request = 0
|
||||
heartbeat = bool(request or (prior_state is not None and prior_state.selfdriveState.enabled))
|
||||
native.set_timer(source_ns // 1000)
|
||||
native.set_aol_test_heartbeat(heartbeat if heartbeat_override is None else heartbeat_override)
|
||||
native.aol_set_host_request(request)
|
||||
frames = classic_frames(card.CI, tick, main=main, lkas=lkas, button=button, engaged=engaged)
|
||||
if drop_wheel:
|
||||
frames = [frame for frame in frames if frame[0] != 0x386]
|
||||
if bad_checksum:
|
||||
frames = [(address, bytes([data[0] ^ 1]) + data[1:] if address == 0x386 else data, bus)
|
||||
for address, data, bus in frames]
|
||||
if inject_relay:
|
||||
camera = next(frame for frame in frames if frame[0] == 0x340 and frame[2] == 2)
|
||||
frames.append((camera[0], camera[1], 0))
|
||||
for address, data, bus in frames:
|
||||
accepted = native.safety_rx_hook(libsafety_py.make_CANPacket(address, bus, data))
|
||||
assert bool(accepted) == (not bad_checksum or address != 0x386), (hex(address), bus)
|
||||
if tick % 100 == 0:
|
||||
native.safety_tick()
|
||||
actual_request = native.aol_get_request_mask()
|
||||
mask = native.aol_get_permission_mask()
|
||||
|
||||
|
||||
|
||||
rx_invalid = bool(native.safety_test_toyota_rx_summary_invalid())
|
||||
relay_fault = bool(native.get_relay_malfunction())
|
||||
assert not relay_fault or inject_relay
|
||||
batch = messaging.log_from_bytes(card_module.can_list_to_can_capnp(frames)).as_builder()
|
||||
batch.logMonoTime = source_ns
|
||||
queued[:] = [batch.to_bytes()]
|
||||
device = messaging.new_message('deviceState', valid=True, logMonoTime=source_ns)
|
||||
device.deviceState.started, device.deviceState.startedMonoTime = True, 9_000_000_000
|
||||
pandas = messaging.new_message('pandaStates', 1, valid=True, logMonoTime=source_ns)
|
||||
panda = pandas.pandaStates[0]
|
||||
panda.safetyModel, panda.safetyParam = config.safetyModel, word
|
||||
panda.alternativeExperience = card.CP.alternativeExperience
|
||||
panda.controlsAllowed = native.get_controls_allowed()
|
||||
panda.safetyRxChecksInvalid = rx_invalid
|
||||
panda.ignitionLine = True
|
||||
panda.faults = (['interruptRateCan2'] if raw_fault else []) + (['relayMalfunction'] if relay_fault else [])
|
||||
slot = PandaSlot(0, 'authority-panda', model, word, card.CP.alternativeExperience,
|
||||
native.get_controls_allowed(), safetyRxChecksInvalid=rx_invalid, faults=raw_fault | int(relay_fault))
|
||||
wire = messaging.new_message('aolSafetyWire', 0, valid=True, logMonoTime=source_ns)
|
||||
wire.aolSafetyWire = encode_safety(SafetyState(1, True, source_ns, source_ns + 30_000_000,
|
||||
model, word, bool(mask & 1), bool(mask & 2), bool(actual_request & 1), bool(actual_request & 2),
|
||||
'authority-panda', card.aol_axis_session(), source_ns, (slot,)))
|
||||
messages = [device.as_reader(), pandas.as_reader(), wire.as_reader()]
|
||||
messages += [previous_sd[name] for name in ('selfdriveState', 'aolAxisState', 'onroadEvents') if name in previous_sd]
|
||||
if 'onroadEvents' not in previous_sd:
|
||||
messages.append(messaging.new_message('onroadEvents', 0, valid=True, logMonoTime=source_ns).as_reader())
|
||||
card.sm.update_msgs(source_ns / 1e9, messages)
|
||||
state, _ = card.state_update()
|
||||
if state.canValid and tick % 2 == 0:
|
||||
assert state.cruiseState.available == bool(main)
|
||||
for service in ('carState', 'carOutput', 'aolIntentWire', 'starpilotCarState'):
|
||||
received.pop(service, None)
|
||||
card.state_publish(state, None)
|
||||
state_envelope = received['carState']
|
||||
output_envelope = received['carOutput']
|
||||
assert state_envelope.logMonoTime == source_ns and state_envelope.valid == state.canValid
|
||||
assert output_envelope.logMonoTime == source_ns
|
||||
|
||||
axis, host, command = drive(tick, intent_message=received['aolIntentWire'],
|
||||
car_state_message=state_envelope, car_output_message=output_envelope,
|
||||
companion=False, authority_message=received.get('starpilotCarState'),
|
||||
allowed=bool(mask & 1), long_allowed=bool(mask & 2), native_request_lat=bool(actual_request & 1),
|
||||
native_request_long=bool(actual_request & 2), panda_controls_allowed=native.get_controls_allowed(),
|
||||
panda_fault=rx_invalid, relay_fault=relay_fault)
|
||||
if sd.conditional_car_state_valid:
|
||||
assert sd.aol_car_state_log_ns == state_envelope.logMonoTime
|
||||
assert controls.sm.logMonoTime['carState'] == state_envelope.logMonoTime
|
||||
assert controls.sm.valid['carState'] == state_envelope.valid
|
||||
assert controls.sm['carState'].to_dict() == state_envelope.carState.to_dict()
|
||||
assert controls.sm.logMonoTime['carOutput'] == output_envelope.logMonoTime
|
||||
assert controls.sm.valid['carOutput'] == output_envelope.valid
|
||||
assert controls.sm['carOutput'].to_dict() == output_envelope.carOutput.to_dict()
|
||||
previous_sd.update(sd.pm.messages)
|
||||
validated = current_native(sd.sm, sd.CP, now_ns=source_ns, axis_session_id=sd.aol_session_id)
|
||||
if not rx_invalid and not relay_fault and freeze_native[0] is None and state.canValid and native.safety_config_valid():
|
||||
assert validated is not None, 'RECORDED_IRQ_ONLY_CURRENT_NATIVE_REJECTED'
|
||||
assert validated is not None and validated.lateralAllowed == bool(mask & 1)
|
||||
assert validated.requestedLateral == bool(actual_request & 1)
|
||||
assert validated.pandaInventory[0].faults == raw_fault
|
||||
assert list(sd.sm['pandaStates'][0].faults) == ([log.PandaState.FaultType.interruptRateCan2] if raw_fault else [])
|
||||
else:
|
||||
assert not axis.lateralActive and not axis.longitudinalActive
|
||||
assert not command.latActive and not command.longActive
|
||||
|
||||
now[0] += 1
|
||||
card.sm.update_msgs(now[0] / 1e9, [controls.pm.messages['carControl']])
|
||||
received.pop('sendcan', None)
|
||||
card.controls_update(state, card.sm['carControl'])
|
||||
steering = []
|
||||
if state.canValid:
|
||||
assert 'sendcan' in received and received['sendcan'].valid
|
||||
steering = [frame for frame in received['sendcan'].sendcan
|
||||
if frame.address == 0x340 and frame.src == 0]
|
||||
assert steering
|
||||
for frame in received['sendcan'].sendcan:
|
||||
accepted = bool(native.safety_tx_hook(libsafety_py.make_CANPacket(frame.address, frame.src, frame.dat)))
|
||||
assert accepted == (not relay_fault)
|
||||
native_counts['accepted' if accepted else 'blocked'] += 1
|
||||
if accepted and frame.address == 0x340 and (int.from_bytes(bytes(frame.dat)[2:4], 'little') & 0x7ff) != 1024:
|
||||
native_counts['nonzero'] += 1
|
||||
native_counts['lateral_only_nonzero'] += int(not native.get_controls_allowed())
|
||||
return state, axis, host, command, steering
|
||||
|
||||
for _ in range(120):
|
||||
step(main=0)
|
||||
assert native.safety_config_valid()
|
||||
for _ in range(2):
|
||||
step(main=1, button=2, engaged=False)
|
||||
for _ in range(30):
|
||||
state, axis, host, command, steering = step(main=1, engaged=True)
|
||||
assert native.get_controls_allowed() and state.cruiseState.enabled
|
||||
assert axis.lateralActive and command.latActive and host.enabled, axis.to_dict()
|
||||
assert not axis.longitudinalActive and not command.longActive
|
||||
nonzero = False
|
||||
for _ in range(30):
|
||||
state, axis, host, command, steering = step(main=1, engaged=True)
|
||||
assert host.enabled and native.get_controls_allowed()
|
||||
assert axis.lateralActive and command.latActive
|
||||
assert not axis.longitudinalActive and not command.longActive
|
||||
nonzero |= abs(command.actuators.torque) > 0
|
||||
assert steering
|
||||
assert nonzero
|
||||
for _ in range(4):
|
||||
step(main=1, engaged=False)
|
||||
for pressed in (1, 0, 1, 0):
|
||||
for _ in range(2):
|
||||
step(main=1, engaged=False, lkas=pressed)
|
||||
for _ in range(20):
|
||||
state, axis, host, command, steering = step(main=1, engaged=False)
|
||||
assert not state.cruiseState.enabled and state.cruiseState.available
|
||||
assert not native.get_controls_allowed() and not host.enabled
|
||||
assert native.aol_get_permission_mask() == 1
|
||||
assert axis.lateralActive and command.latActive, (axis.to_dict(), native.aol_get_permission_mask())
|
||||
assert not axis.longitudinalActive and not command.longActive
|
||||
assert diagnostic_requests == []
|
||||
assert card.ci_initialized and saved.get_bool('ControlsReady')
|
||||
if denial == 'stale-wire':
|
||||
freeze_native[0] = last_native[0]
|
||||
telemetry_rate_bad[0] = denial == 'telemetry-rate'
|
||||
for _ in range(120 if denial in ('frequency', 'telemetry-rate') else 30 if denial == 'stale-wire' else 4):
|
||||
state, axis, host, command, steering = step(
|
||||
main=0 if denial == 'request' else 1, engaged=False,
|
||||
withdraw_request=denial == 'request', bad_checksum=denial == 'checksum',
|
||||
heartbeat_override=False if denial == 'heartbeat' else None,
|
||||
inject_relay=denial == 'relay', drop_wheel=denial == 'frequency')
|
||||
assert not axis.lateralActive and not command.latActive
|
||||
assert not axis.longitudinalActive and not command.longActive
|
||||
assert native.aol_get_permission_mask() == (1 if denial == 'telemetry-rate' else 0)
|
||||
if denial == 'telemetry-rate':
|
||||
assert native.safety_config_valid() and not sd.sm.freq_ok['pandaStates']
|
||||
assert all((int.from_bytes(bytes(frame.dat)[2:4], 'little') & 0x7ff) == 1024 for frame in steering)
|
||||
assert native_counts['nonzero'] > 0 and native_counts['lateral_only_nonzero'] > 0
|
||||
print(json.dumps({'raw_fault': raw_fault, 'denial': denial, 'native_counts': native_counts}))
|
||||
finally:
|
||||
patch_scope.__exit__(None, None, None)
|
||||
finally:
|
||||
native.init_tests()
|
||||
native.set_alternative_experience(0)
|
||||
native.set_safety_hooks(structs.CarParams.SafetyModel.noOutput, 0)
|
||||
case.doCleanups()
|
||||
if card is not None:
|
||||
card.vehicle_startup.close()
|
||||
@@ -0,0 +1,75 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from opendbc.car import gen_empty_fingerprint
|
||||
from opendbc.car.hyundai.interface import CarInterface
|
||||
from opendbc.car.hyundai.values import CAR, HyundaiFlags
|
||||
from openpilot.cereal import log
|
||||
from openpilot.starpilot.aol.panda_faults import blocking_fault_list, blocking_fault_mask
|
||||
|
||||
|
||||
def recorded():
|
||||
fp = gen_empty_fingerprint()
|
||||
fp[0].update({0x391: 8, 0x38d: 8})
|
||||
fp[2][0x485] = 4
|
||||
fp[1][0x500] = 8
|
||||
cp = CarInterface.get_params(CAR.HYUNDAI_SONATA_HYBRID, fp, [], False, False, False)
|
||||
cp.safetyConfigs[0].safetyParam |= 0xC00
|
||||
cp.alternativeExperience = 32
|
||||
assert cp.flags == 0x1011580 and cp.safetyConfigs[0].safetyParam == 0xC02
|
||||
return cp
|
||||
|
||||
|
||||
@pytest.mark.parametrize('mask', [1, 2, 4, 16, 32, 64, 256, 1 << 18, 9, 10, 12, 24, 0xffffffff, -1, 1 << 32, None, 'invalid'])
|
||||
def test_other_mixed_or_malformed_faults_remain_blocking(mask):
|
||||
assert blocking_fault_mask(mask, recorded())
|
||||
|
||||
|
||||
@pytest.mark.parametrize('field,value', [('brand', 'ford'), ('carFingerprint', CAR.HYUNDAI_SONATA),
|
||||
('flags', 0x1011580 ^ 0x80), ('alternativeExperience', 0), ('alternativeExperience', 1),
|
||||
('alternativeExperience', 33), ('passive', True), ('dashcamOnly', True), ('notCar', True),
|
||||
('openpilotLongitudinalControl', True), ('pcmCruise', False)])
|
||||
def test_unrecorded_configuration_does_not_inherit_diagnostic_classification(field, value):
|
||||
cp = recorded()
|
||||
setattr(cp, field, value)
|
||||
assert blocking_fault_mask(8, cp)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('word', [2, 0x402, 0xC00, 0xC01, 0xC04, 0xC0A, 0x840A])
|
||||
def test_other_native_profiles_remain_blocking(word):
|
||||
cp = recorded()
|
||||
cp.safetyConfigs[0].safetyParam = word
|
||||
assert blocking_fault_mask(8, cp)
|
||||
|
||||
|
||||
def test_raw_fault_list_and_carparams_are_retained():
|
||||
cp = recorded()
|
||||
before = cp.to_dict()
|
||||
faults = [log.PandaState.FaultType.interruptRateCan2]
|
||||
assert not blocking_fault_mask(0, cp)
|
||||
assert not blocking_fault_mask(8, cp)
|
||||
assert not blocking_fault_list(faults, cp)
|
||||
assert faults == [log.PandaState.FaultType.interruptRateCan2]
|
||||
for fault in (log.PandaState.FaultType.relayMalfunction, log.PandaState.FaultType.interruptRateCan1,
|
||||
log.PandaState.FaultType.registerDivergent):
|
||||
assert blocking_fault_list([*faults, fault], cp)
|
||||
assert cp.to_dict() == before
|
||||
|
||||
|
||||
def test_extra_owner_and_legacy_model_remain_blocking():
|
||||
cp = recorded()
|
||||
cp.safetyConfigs[0].safetyModel = 'hyundaiLegacy'
|
||||
assert blocking_fault_mask(8, cp)
|
||||
cp = recorded()
|
||||
cp.safetyConfigs = [cp.safetyConfigs[0].to_dict(), cp.safetyConfigs[0].to_dict()]
|
||||
assert blocking_fault_mask(8, cp)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('identity', [CAR.HYUNDAI_IONIQ_6, CAR.HYUNDAI_SONATA])
|
||||
def test_unrelated_hyundai_skips_flag_construction_and_qualification(identity):
|
||||
cp = recorded()
|
||||
cp.carFingerprint = identity
|
||||
with patch.object(HyundaiFlags, '__or__', side_effect=AssertionError('unrelated flag work')), \
|
||||
patch('opendbc.car.hyundai.classic_scc_aol.qualified', side_effect=AssertionError('unrelated qualification')):
|
||||
assert blocking_fault_mask(8, cp)
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"protected_inventory_sha256": "62a71c3fb7b92b3641b60dd216962d3fca64bdfb459fd306ef502470b47541e3", "schema_version": 1}
|
||||
{"protected_inventory_sha256": "c6b63f02f9d83666c769db699ce400b40c5cebc3b53795d5d8bae9d880a8d4bd", "schema_version": 1}
|
||||
|
||||
@@ -33,6 +33,8 @@ DEBUG_SAFETY_CLASS_TARGETS = {
|
||||
|
||||
|
||||
RELEASE_TARGETS = (
|
||||
"openpilot/starpilot/aol/tests/test_sonata_irq_callers.py",
|
||||
"openpilot/starpilot/aol/tests/test_sonata_irq_policy.py",
|
||||
"opendbc_repo/opendbc/car/hyundai/tests/test_classic_aol_dash.py",
|
||||
"opendbc_repo/opendbc/car/hyundai/tests/test_g90_continuity.py",
|
||||
"opendbc_repo/opendbc/car/hyundai/tests/test_disable_response.py",
|
||||
|
||||
Reference in New Issue
Block a user