mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-28 00:53:45 +08:00
Keep Ioniq 5 PE READY startup stock until ECU takeover is proven
This commit is contained in:
firestar5683
@@ -385,9 +385,6 @@ class CarInterface(CarInterfaceBase):
|
||||
self.CS.forte_lkas_sources.update(can_packets)
|
||||
ret = super().update(can_packets)
|
||||
self.CS.finalize_blended_cancel(ret)
|
||||
pe_startup = getattr(self.CC, 'pe_startup', None)
|
||||
if pe_startup is not None:
|
||||
pe_startup.observe_parked_ready_can(can_packets)
|
||||
return ret
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -1,76 +0,0 @@
|
||||
from copy import deepcopy
|
||||
from math import isfinite
|
||||
|
||||
from opendbc.can.parser import MAX_BAD_COUNTER
|
||||
from opendbc.car import Bus, structs
|
||||
from opendbc.car.hyundai.ccnc_ev_stock import qualified as stock_qualified
|
||||
from opendbc.car.hyundai.ioniq5pe_ready import PEReadySource
|
||||
from opendbc.car.hyundai.values import CAR
|
||||
|
||||
|
||||
class PEParkedReadyProof:
|
||||
def __init__(self, stock_cp, floor_ns):
|
||||
if (stock_cp.carFingerprint != CAR.HYUNDAI_IONIQ_5_PE or not stock_qualified(stock_cp) or
|
||||
not stock_cp.alphaLongitudinalAvailable or floor_ns <= 0):
|
||||
raise ValueError('Exact unmarked PE stock configuration and source floor required')
|
||||
from opendbc.car.hyundai.interface import CarInterface
|
||||
self.ci = CarInterface(stock_cp.as_reader().as_builder())
|
||||
self.snapshot = deepcopy(self.ci.CP.to_dict())
|
||||
self.floor_ns = floor_ns
|
||||
self.last_stamp = 0
|
||||
self.invalid = False
|
||||
self.ready = PEReadySource()
|
||||
|
||||
def observe(self, packets, now_ns):
|
||||
for stamp, frames in packets:
|
||||
if type(stamp) is not int or stamp <= 0 or stamp > now_ns:
|
||||
self.invalid = True
|
||||
continue
|
||||
if stamp <= self.floor_ns:
|
||||
if self.last_stamp:
|
||||
self.invalid = True
|
||||
continue
|
||||
if stamp <= self.last_stamp:
|
||||
self.invalid = True
|
||||
continue
|
||||
self.last_stamp = stamp
|
||||
self.ready.update([(stamp, frames)], now_ns)
|
||||
self.ci.update([(stamp, frames)])
|
||||
|
||||
def current(self, now_ns):
|
||||
state = self.ci.CS.out
|
||||
if (self.invalid or self.ci.CP.to_dict() != self.snapshot or not state.canValid or state.canTimeout or
|
||||
state.gearShifter != structs.CarState.GearShifter.park or not state.standstill or
|
||||
not all(isfinite(value) for value in (state.vEgo, state.vEgoRaw)) or
|
||||
state.gasPressed or state.regenBraking or state.steerFaultTemporary or state.steerFaultPermanent or state.accFaulted or
|
||||
state.doorOpen or state.seatbeltUnlatched or state.cruiseState.enabled or not state.cruiseState.available or
|
||||
self.ready.current(now_ns, self.floor_ns, 100_000_000) is not True):
|
||||
return False
|
||||
parser = self.ci.can_parsers[Bus.pt]
|
||||
camera = self.ci.can_parsers[Bus.cam]
|
||||
if parser.bus != 1 or camera.bus != 2 or not camera.can_valid or camera.bus_timeout:
|
||||
return False
|
||||
names = ('MDPS', 'ACCELERATOR', 'WHEEL_SPEEDS', 'TCS', 'STEERING_SENSORS', 'CRUISE_BUTTONS',
|
||||
'SCC_CONTROL', 'DOORS_SEATBELTS')
|
||||
for source_parser, name in [*((parser, name) for name in names), (camera, 'CAM_0x362')]:
|
||||
source = source_parser.message_states.get(source_parser.dbc.name_to_msg[name].address)
|
||||
if (source is None or not source.timestamps or source.counter_fail >= MAX_BAD_COUNTER or
|
||||
not self.floor_ns < int(source.timestamps[-1]) <= now_ns or
|
||||
now_ns - int(source.timestamps[-1]) > source.timeout_threshold):
|
||||
return False
|
||||
buttons = parser.vl['CRUISE_BUTTONS']
|
||||
return all(buttons[name] == 0 for name in ('CRUISE_BUTTONS', 'ADAPTIVE_CRUISE_MAIN_BTN', 'NORMAL_CRUISE_MAIN_BTN'))
|
||||
|
||||
def matches_target(self, ci, now_ns):
|
||||
if not self.current(now_ns) or ci.CP.carFingerprint != CAR.HYUNDAI_IONIQ_5_PE:
|
||||
return False
|
||||
stock, target = self.ci.can_parsers[Bus.pt], ci.can_parsers[Bus.pt]
|
||||
if target.bus != stock.bus or not ci.CS.out.canValid or ci.CS.out.canTimeout:
|
||||
return False
|
||||
for name in ('MDPS', 'ACCELERATOR', 'WHEEL_SPEEDS', 'TCS', 'STEERING_SENSORS', 'CRUISE_BUTTONS'):
|
||||
address = stock.dbc.name_to_msg[name].address
|
||||
a, b = stock.message_states.get(address), target.message_states.get(address)
|
||||
if a is None or b is None or not a.timestamps or not b.timestamps or a.timestamps[-1] != b.timestamps[-1]:
|
||||
return False
|
||||
return bool(ci.CS.out.gearShifter == structs.CarState.GearShifter.park and ci.CS.out.standstill and
|
||||
not ci.CS.out.gasPressed and not ci.CS.out.regenBraking)
|
||||
@@ -68,7 +68,6 @@ class PEPendingStartup:
|
||||
self.native_context = None
|
||||
self.emission_current = False
|
||||
self.quiet_failure = None
|
||||
self.parked_ready = None
|
||||
|
||||
def _ready_probe(self):
|
||||
observer = PEReadySource()
|
||||
@@ -87,32 +86,13 @@ class PEPendingStartup:
|
||||
|
||||
return observer.current(self.boot_clock(), start, 100_000_000)
|
||||
|
||||
def _parked_ready_probe(self):
|
||||
from opendbc.car.hyundai.ioniq5pe_parked_ready import PEParkedReadyProof
|
||||
observer = PEParkedReadyProof(self.stock_cp, self.boot_clock())
|
||||
deadline = self.clock() + .5
|
||||
while self.clock() < deadline:
|
||||
packets = self.callbacks[0](wait_for_one=False)
|
||||
stamped = [(int(getattr(packet, 'log_mono_time_ns', 0)), list(packet)) for packet in packets]
|
||||
observer.observe(stamped, self.boot_clock())
|
||||
if observer.current(self.boot_clock()):
|
||||
return observer
|
||||
time.sleep(.005)
|
||||
return None
|
||||
|
||||
def observe_parked_ready_can(self, packets):
|
||||
if self.pending_target and not self.session_attempted and self.parked_ready is not None:
|
||||
self.parked_ready.observe(packets, self.boot_clock())
|
||||
|
||||
def prepare(self, *, admission):
|
||||
if self.owner.phase is not Phase.NEW:
|
||||
raise RuntimeError('PE preparation is single-use')
|
||||
self.diagnostic_admission = admission
|
||||
self.prepared_floor_ns = self.boot_clock()
|
||||
ready = self._ready_probe() if admission() is True and self.prepared_floor_ns > 0 else None
|
||||
if ready is True:
|
||||
self.parked_ready = self._parked_ready_probe()
|
||||
if ready is not False and self.parked_ready is None:
|
||||
if ready is not False:
|
||||
self.owner.phase = Phase.STOCK
|
||||
self.owner.cp = copy_cp(self.stock_cp)
|
||||
else:
|
||||
@@ -230,11 +210,9 @@ class PEPendingStartup:
|
||||
parser = ci.can_parsers[Bus.pt]
|
||||
accelerator = parser.message_states[parser.dbc.name_to_msg['ACCELERATOR'].address]
|
||||
ready = ci.CS.pe_ready.current(self.boot_clock(), self.prepared_floor_ns, accelerator.timeout_threshold)
|
||||
ready_allowed = (ready is False if self.parked_ready is None else
|
||||
ready is True and self.parked_ready.matches_target(ci, self.boot_clock()))
|
||||
if not ready_allowed:
|
||||
if ready is not False:
|
||||
self.pending_target = False
|
||||
self.fault = 'PE startup source/READY/park proof lost before takeover; no ECU mutation'
|
||||
self.fault = 'PE Ready/unknown before takeover; abort published target without ECU mutation'
|
||||
self.owner.phase = Phase.ABORT
|
||||
raise RuntimeError(self.fault)
|
||||
try:
|
||||
@@ -249,14 +227,6 @@ class PEPendingStartup:
|
||||
ack_floor_ns = self.boot_clock()
|
||||
if not self._confirm_quiet(ci, ack_floor_ns):
|
||||
raise RuntimeError(f'PE acknowledged request did not establish quiet/current ownership: {self.quiet_failure}')
|
||||
if self.parked_ready is not None:
|
||||
state = ci.CS.out
|
||||
from opendbc.car import structs
|
||||
if (state.gearShifter != structs.CarState.GearShifter.park or not state.standstill or
|
||||
state.gasPressed or state.regenBraking or state.doorOpen or state.seatbeltUnlatched or
|
||||
state.steerFaultTemporary or state.steerFaultPermanent or state.accFaulted or
|
||||
ci.CS.pe_ready.current(self.boot_clock(), ack_floor_ns, accelerator.timeout_threshold) is not True):
|
||||
raise RuntimeError('PE parked READY state lost during acknowledged takeover')
|
||||
self.owner.phase = Phase.OWNED
|
||||
self.source_floor_ns = self.boot_clock()
|
||||
self._tester()
|
||||
|
||||
@@ -1,224 +0,0 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from opendbc.can import CANPacker
|
||||
from opendbc.car import Bus, structs
|
||||
from opendbc.car.hyundai.interface import CarInterface
|
||||
from opendbc.car.hyundai.ioniq5pe_long_profile import candidate
|
||||
from opendbc.car.hyundai.ioniq5pe_startup import PEPendingStartup, Phase
|
||||
from opendbc.car.hyundai.ioniq6_handoff import TimestampedCanPacket
|
||||
from opendbc.car.hyundai.hyundaicanfd import hkg_can_fd_checksum
|
||||
from opendbc.car.hyundai.tests.test_ioniq5pe_pending_startup import ThreadFixture
|
||||
from opendbc.car.hyundai.tests.test_ioniq5pe_stock import params
|
||||
from opendbc.car.hyundai.tests.test_ev9_long_controller import packets
|
||||
from opendbc.car.hyundai.values import CAR, DBC
|
||||
|
||||
|
||||
class ParkedFixture:
|
||||
def __init__(self):
|
||||
self.ns = 1_000_000_000
|
||||
self.tick = 0
|
||||
self.packer = CANPacker(DBC[CAR.HYUNDAI_IONIQ_5_PE][Bus.pt])
|
||||
self.queries = []
|
||||
self.sent = []
|
||||
self.gear = 0
|
||||
self.moving = False
|
||||
self.ready = True
|
||||
self.cruise = False
|
||||
self.button = 0
|
||||
self.main = False
|
||||
self.brake = False
|
||||
self.stock_scc = True
|
||||
self.quiet = False
|
||||
self.disable_ok = True
|
||||
self.restore_ok = True
|
||||
self.drop = None
|
||||
self.stock = params()
|
||||
self.owner = PEPendingStartup(candidate(self.stock, enabled=True, is_release=False),
|
||||
(self.recv, self.sent.extend), stock_cp=self.stock,
|
||||
clock=lambda: self.ns / 1e9, boot_clock=lambda: self.ns,
|
||||
exchange=self.exchange, thread_factory=ThreadFixture)
|
||||
|
||||
def frames(self):
|
||||
frames = packets(self.packer, self.tick, advance_counter=True, gear=self.gear, moving=self.moving,
|
||||
cruise=self.cruise, cruise_button=self.button, adaptive_main=self.main)
|
||||
result = []
|
||||
for address, data, bus in frames:
|
||||
if address == 0x35:
|
||||
raw = bytearray(data)
|
||||
if self.ready:
|
||||
raw[3] |= 0x40
|
||||
raw[:2] = hkg_can_fd_checksum(address, None, raw).to_bytes(2, 'little')
|
||||
data = bytes(raw)
|
||||
if address == 0x175 and self.brake:
|
||||
_, data, _ = self.packer.make_can_msg('TCS', 1, {'COUNTER': self.tick // 2, 'DriverBraking': 1})
|
||||
if address == 0x362:
|
||||
_, data, _ = self.packer.make_can_msg('CAM_0x362', 2, {'COUNTER': self.tick // 5})
|
||||
result.append((address, data, bus))
|
||||
if self.stock_scc and not self.quiet and self.tick % 2 == 0:
|
||||
result.append(self.packer.make_can_msg('SCC_CONTROL', 1, {'COUNTER': self.tick // 2, 'ACCMode': int(self.cruise)}))
|
||||
return [frame for frame in result if frame[0] != self.drop]
|
||||
|
||||
def recv(self, **kwargs):
|
||||
self.tick += 1
|
||||
self.ns += 10_000_000
|
||||
return [TimestampedCanPacket(self.frames(), self.ns)]
|
||||
|
||||
def exchange(self, recv, send, request, response):
|
||||
self.queries.append((request, response))
|
||||
if request == b'\x28\x01\x01':
|
||||
if self.disable_ok:
|
||||
self.quiet = True
|
||||
return self.disable_ok
|
||||
if request == b'\x28\x00\x01':
|
||||
return self.restore_ok
|
||||
return True
|
||||
|
||||
def prepare(self):
|
||||
cp = self.owner.prepare(admission=lambda: True)
|
||||
self.ci = CarInterface(cp)
|
||||
self.owner.configure(self.ci)
|
||||
self.owner.finalize_aol_configuration(self.ci)
|
||||
self.owner.seal_publication()
|
||||
return cp
|
||||
|
||||
def warm(self, ticks=15):
|
||||
for _ in range(ticks):
|
||||
packet = self.recv()[0]
|
||||
self.ci.update([(packet.log_mono_time_ns, list(packet))])
|
||||
|
||||
def commit(self):
|
||||
self.owner.commit_pending(self.ci, sources_current=self.owner.sources_current(self.ci, self.ns), control_current=True)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def parked():
|
||||
value = ParkedFixture()
|
||||
with patch('time.clock_gettime_ns', side_effect=lambda clock: value.ns):
|
||||
yield value
|
||||
value.owner.close()
|
||||
|
||||
|
||||
def test_ready_parked_publication_is_pending_and_sends_nothing_before_initialized_commit(parked):
|
||||
stock = parked.stock.to_dict()
|
||||
cp = parked.prepare()
|
||||
assert cp.openpilotLongitudinalControl and cp.safetyConfigs[0].safetyParam == 0x5495
|
||||
assert parked.owner.parked_ready is not None and parked.owner.pending_target
|
||||
assert parked.owner.owner.phase is Phase.PENDING and not parked.owner.owner.active()
|
||||
assert not parked.queries and not parked.sent and parked.owner.thread is None
|
||||
assert parked.stock.to_dict() == stock
|
||||
parked.warm()
|
||||
assert parked.owner.parked_ready.matches_target(parked.ci, parked.ns)
|
||||
cc = structs.CarControl(enabled=True, latActive=True, longActive=True)
|
||||
assert parked.ci.apply(cc.as_reader(), parked.ns)[1] == []
|
||||
parked.commit()
|
||||
assert parked.owner.owner.active() and parked.owner.thread.started
|
||||
assert parked.queries[:2] == [(b'\x10\x03', b'\x50\x03'), (b'\x28\x01\x01', b'\x68\x01')]
|
||||
assert not parked.owner.can_emit(parked.ci.CC) and not parked.owner.handed_off
|
||||
assert not parked.owner.sources_current(parked.ci, parked.ns)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('failure', ['drive', 'reverse', 'neutral', 'moving', 'not_ready', 'cruise', 'cancel', 'main',
|
||||
'missing_scc', 'missing_cam', 'missing_mdps', 'missing_accel', 'missing_wheels',
|
||||
'missing_tcs', 'missing_steering', 'missing_buttons', 'missing_doors'])
|
||||
def test_unproven_ready_start_preserves_exact_stock_and_never_disables(parked, failure):
|
||||
if failure in ('drive', 'reverse', 'neutral'):
|
||||
parked.gear = {'drive': 5, 'reverse': 7, 'neutral': 6}[failure]
|
||||
elif failure == 'moving':
|
||||
parked.moving = True
|
||||
elif failure == 'not_ready':
|
||||
# The existing READY=false path remains separate and unchanged.
|
||||
with patch.object(parked.owner, '_ready_probe', return_value=None):
|
||||
parked.prepare()
|
||||
assert parked.owner.owner.phase is Phase.STOCK
|
||||
return
|
||||
elif failure == 'cruise':
|
||||
parked.cruise = True
|
||||
elif failure == 'cancel':
|
||||
parked.button = 4
|
||||
elif failure == 'main':
|
||||
parked.main = True
|
||||
else:
|
||||
parked.drop = {'missing_scc': 0x1A0, 'missing_cam': 0x362, 'missing_mdps': 0xEA, 'missing_accel': 0x35,
|
||||
'missing_wheels': 0xA0, 'missing_tcs': 0x175, 'missing_steering': 0x125,
|
||||
'missing_buttons': 0x1CF, 'missing_doors': 0x411}[failure]
|
||||
cp = parked.prepare()
|
||||
assert cp.to_dict() == parked.stock.to_dict()
|
||||
assert parked.owner.owner.phase is Phase.STOCK and not parked.owner.pending_target
|
||||
assert not parked.queries and not parked.sent
|
||||
|
||||
|
||||
@pytest.mark.parametrize('failure', ['drive', 'reverse', 'neutral', 'moving', 'not_ready', 'cruise', 'cancel', 'main',
|
||||
'stale', 'lost_proof', 'target_stamp', 'gas', 'door', 'seatbelt', 'fault',
|
||||
'bad_counter', 'cp_mutation', 'reordered', 'future', 'zero_stamp'])
|
||||
def test_lost_parked_proof_before_commit_aborts_without_ecu_mutation(parked, failure):
|
||||
parked.prepare()
|
||||
parked.warm()
|
||||
assert parked.owner.parked_ready.matches_target(parked.ci, parked.ns)
|
||||
if failure in ('drive', 'reverse', 'neutral'):
|
||||
parked.gear = {'drive': 5, 'reverse': 7, 'neutral': 6}[failure]
|
||||
parked.warm()
|
||||
elif failure in ('moving', 'cruise'):
|
||||
setattr(parked, failure, True)
|
||||
parked.warm()
|
||||
elif failure == 'not_ready':
|
||||
parked.ready = False
|
||||
parked.warm()
|
||||
elif failure in ('cancel', 'main'):
|
||||
parked.button, parked.main = (4, False) if failure == 'cancel' else (0, True)
|
||||
parked.warm()
|
||||
else:
|
||||
proof = parked.owner.parked_ready
|
||||
if failure == 'stale':
|
||||
parked.ns += 300_000_001
|
||||
elif failure == 'lost_proof':
|
||||
proof.invalid = True
|
||||
elif failure == 'target_stamp':
|
||||
source = parked.ci.can_parsers[Bus.pt].message_states[0x35]
|
||||
source.timestamps[-1] -= 1
|
||||
elif failure in ('gas', 'door', 'seatbelt', 'fault'):
|
||||
setattr(proof.ci.CS.out, {'gas': 'gasPressed', 'door': 'doorOpen', 'seatbelt': 'seatbeltUnlatched',
|
||||
'fault': 'steerFaultTemporary'}[failure], True)
|
||||
elif failure == 'bad_counter':
|
||||
from opendbc.can.parser import MAX_BAD_COUNTER
|
||||
proof.ci.can_parsers[Bus.pt].message_states[0x35].counter_fail = MAX_BAD_COUNTER
|
||||
elif failure == 'cp_mutation':
|
||||
proof.ci.CP.pcmCruise = False
|
||||
elif failure == 'reordered':
|
||||
proof.observe([(proof.last_stamp, [])], parked.ns)
|
||||
elif failure == 'future':
|
||||
proof.observe([(parked.ns + 1, [])], parked.ns)
|
||||
elif failure == 'zero_stamp':
|
||||
proof.observe([(0, [])], parked.ns)
|
||||
with pytest.raises(RuntimeError):
|
||||
parked.commit()
|
||||
assert not parked.queries and not parked.owner.session_attempted and not parked.owner.takeover_attempted
|
||||
assert not parked.owner.can_emit(parked.ci.CC)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('failure', ['disable_negative', 'quiet_collision', 'movement_during_query', 'ready_loss_during_query'])
|
||||
def test_acknowledgement_quiet_and_parked_continuity_are_required(parked, failure):
|
||||
parked.prepare()
|
||||
parked.warm()
|
||||
original = parked.owner.exchange
|
||||
|
||||
def exchange(*args):
|
||||
result = original(*args)
|
||||
if args[2] == b'\x28\x01\x01':
|
||||
if failure == 'quiet_collision':
|
||||
parked.quiet = False
|
||||
elif failure == 'movement_during_query':
|
||||
parked.gear, parked.moving = 5, True
|
||||
elif failure == 'ready_loss_during_query':
|
||||
parked.ready = False
|
||||
return result
|
||||
parked.owner.exchange = exchange
|
||||
parked.disable_ok = failure != 'disable_negative'
|
||||
before = parked.ci.CP.to_dict()
|
||||
with pytest.raises(RuntimeError):
|
||||
parked.commit()
|
||||
assert parked.owner.takeover_attempted and parked.owner.restored
|
||||
assert parked.owner.owner.phase is Phase.RESTORED
|
||||
assert parked.ci.CP.to_dict() == before
|
||||
assert not parked.owner.can_emit(parked.ci.CC)
|
||||
@@ -1,86 +0,0 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from opendbc.car import structs
|
||||
from opendbc.car.hyundai.tests.test_ioniq5pe_parked_ready import ParkedFixture
|
||||
from opendbc.safety.tests.libsafety import libsafety_py
|
||||
|
||||
|
||||
@pytest.mark.parametrize('withdrawal', ['cancel', 'main', 'brake'])
|
||||
def test_parked_ready_pending_neutral_native_handoff_requires_new_set_then_obeys_withdrawal(withdrawal):
|
||||
fixture = ParkedFixture()
|
||||
native = libsafety_py.libsafety
|
||||
native.set_alternative_experience(0)
|
||||
assert native.set_safety_hooks(structs.CarParams.SafetyModel.elm327, 1) == 0
|
||||
native.init_tests()
|
||||
|
||||
def packet(frame):
|
||||
return libsafety_py.make_CANPacket(frame[0], frame[2], frame[1])
|
||||
with patch('time.clock_gettime_ns', side_effect=lambda clock: fixture.ns):
|
||||
try:
|
||||
fixture.prepare()
|
||||
fixture.warm()
|
||||
fixture.commit()
|
||||
assert native.set_safety_hooks(structs.CarParams.SafetyModel.hyundaiCanfd, 0x5495) == 0
|
||||
|
||||
def tick():
|
||||
batch = fixture.recv()[0]
|
||||
native.set_timer(fixture.ns // 1000)
|
||||
for frame in batch:
|
||||
accepted = native.safety_rx_hook(packet(frame))
|
||||
if frame[0] in (0x35, 0x175, 0xA0, 0xEA, 0x1CF):
|
||||
assert accepted, hex(frame[0])
|
||||
fixture.ci.update([(batch.log_mono_time_ns, list(batch))])
|
||||
cfg = fixture.ci.CP.safetyConfigs[0]
|
||||
panda = SimpleNamespace(safetyModel=cfg.safetyModel, safetyParam=cfg.safetyParam,
|
||||
alternativeExperience=0, faults=[], safetyRxChecksInvalid=False,
|
||||
ignitionLine=True, ignitionCan=False)
|
||||
fixture.owner.after_state(ci=fixture.ci, state=fixture.ci.CS.out, now_ns=fixture.ns,
|
||||
control_enabled=False, control_current=True, pandas=[panda],
|
||||
panda_log_ns=fixture.ns, panda_recv_ns=fixture.ns, panda_current=True)
|
||||
for _ in range(15):
|
||||
tick()
|
||||
assert not native.get_controls_allowed()
|
||||
cc = structs.CarControl(enabled=True, latActive=True, longActive=True)
|
||||
cc.actuators.accel = 1.
|
||||
cc.actuators.steeringAngleDeg = 1.
|
||||
cc.actuators.longControlState = structs.CarControl.Actuators.LongControlState.pid
|
||||
_, neutral = fixture.ci.apply(cc.as_reader(), fixture.ns)
|
||||
assert {0x1A0, 0xCB, 0x100, 0x730}.issubset({frame[0] for frame in neutral})
|
||||
assert ((next(frame[1] for frame in neutral if frame[0] == 0x1A0)[8] >> 4) & 7) == 0
|
||||
assert ((next(frame[1] for frame in neutral if frame[0] == 0xCB)[3] >> 4) & 15) == 1
|
||||
for frame in neutral:
|
||||
assert native.safety_tx_hook(packet(frame)), hex(frame[0])
|
||||
fixture.owner.sent(neutral, valid=True)
|
||||
assert fixture.owner.handed_off and fixture.owner.thread.joined
|
||||
fixture.gear, fixture.moving = 5, True
|
||||
for _ in range(15):
|
||||
tick()
|
||||
assert not native.get_controls_allowed()
|
||||
denied = fixture.packer.make_can_msg('SCC_CONTROL', 1, {'ACCMode': 1, 'aReqRaw': 1., 'aReqValue': 1.})
|
||||
assert not native.safety_tx_hook(packet(denied))
|
||||
fixture.button = 2
|
||||
for _ in range(2):
|
||||
tick()
|
||||
fixture.button = 0
|
||||
for _ in range(2):
|
||||
tick()
|
||||
assert native.get_controls_allowed()
|
||||
assert native.safety_tx_hook(packet(denied))
|
||||
_, active = fixture.ci.apply(cc.as_reader(), fixture.ns)
|
||||
for frame in active:
|
||||
assert native.safety_tx_hook(packet(frame)), hex(frame[0])
|
||||
if withdrawal == 'cancel':
|
||||
fixture.button = 4
|
||||
elif withdrawal == 'main':
|
||||
fixture.main = True
|
||||
else:
|
||||
fixture.brake = True
|
||||
for _ in range(2):
|
||||
tick()
|
||||
assert not native.get_controls_allowed()
|
||||
assert not native.safety_tx_hook(packet(denied))
|
||||
finally:
|
||||
fixture.owner.close()
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"protected_inventory_sha256": "c682d5e7e787cfc45b49426adfbc9c5d9825d7be55b9d87e79ae62807f6c1690", "schema_version": 1}
|
||||
{"protected_inventory_sha256": "0e6a3515188d9d8b95b31741e53a436e29396874708fb93accd32a9d0d7f4b5e", "schema_version": 1}
|
||||
|
||||
@@ -33,7 +33,6 @@ DEBUG_SAFETY_CLASS_TARGETS = {
|
||||
|
||||
|
||||
RELEASE_TARGETS = (
|
||||
"opendbc_repo/opendbc/car/hyundai/tests/test_ioniq5pe_parked_ready.py",
|
||||
"opendbc_repo/opendbc/car/hyundai/tests/test_radar_hda2_startup.py",
|
||||
"opendbc_repo/opendbc/car/hyundai/tests/test_radar_hda2_startup_guards.py",
|
||||
"opendbc_repo/opendbc/car/hyundai/tests/test_radar_hda2_startup_lifecycle.py::test_exact_factory_stock_copy_and_owner_preserve_all_params",
|
||||
|
||||
Reference in New Issue
Block a user