From c68fdb0bc67d30ea30bbbbf3be13f30b9e9e1429 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Fri, 25 Sep 2026 14:53:18 -0500 Subject: [PATCH] Vehicle Preferences Resolve startup preferences and controller extensions through capability-aware vehicle owners. --- .../car/gm/tests/test_gm_auto_identity.py | 174 +++++++++ .../test_blended_longitudinal_lifecycle.py | 331 ++++++++++++++++++ .../car/hyundai/tests/test_g90_startup.py | 190 ++++++++++ .../tests/test_hyundai_blended_cruise_ack.py | 104 ++++++ .../car/tesla/tests/test_screen_intent.py | 115 ++++++ openpilot/starpilot/controller_extensions.py | 106 ++++++ .../lateral/tests/test_turn_assist.py | 113 ++++++ .../tests/test_controller_extensions.py | 135 +++++++ openpilot/starpilot/tests/test_ev6_startup.py | 194 ++++++++++ .../starpilot/tests/test_g90_lead_inputs.py | 169 +++++++++ .../tests/test_gv70_fallback_cancel.py | 30 ++ .../starpilot/tests/test_gv70_lead_inputs.py | 192 ++++++++++ .../starpilot/tests/test_gv70_startup.py | 89 +++++ .../tests/test_vehicle_preferences.py | 124 +++++++ .../tests/test_vehicle_state_context.py | 98 ++++++ openpilot/starpilot/vehicle_preferences.py | 203 +++++++++++ openpilot/starpilot/vehicle_startup.py | 69 ++++ 17 files changed, 2436 insertions(+) create mode 100644 opendbc_repo/opendbc/car/gm/tests/test_gm_auto_identity.py create mode 100644 opendbc_repo/opendbc/car/hyundai/tests/test_blended_longitudinal_lifecycle.py create mode 100644 opendbc_repo/opendbc/car/hyundai/tests/test_g90_startup.py create mode 100644 opendbc_repo/opendbc/safety/tests/test_hyundai_blended_cruise_ack.py create mode 100644 openpilot/starpilot/car/tesla/tests/test_screen_intent.py create mode 100644 openpilot/starpilot/controller_extensions.py create mode 100644 openpilot/starpilot/lateral/tests/test_turn_assist.py create mode 100644 openpilot/starpilot/tests/test_controller_extensions.py create mode 100644 openpilot/starpilot/tests/test_ev6_startup.py create mode 100644 openpilot/starpilot/tests/test_g90_lead_inputs.py create mode 100644 openpilot/starpilot/tests/test_gv70_fallback_cancel.py create mode 100644 openpilot/starpilot/tests/test_gv70_lead_inputs.py create mode 100644 openpilot/starpilot/tests/test_gv70_startup.py create mode 100644 openpilot/starpilot/tests/test_vehicle_preferences.py create mode 100644 openpilot/starpilot/tests/test_vehicle_state_context.py create mode 100644 openpilot/starpilot/vehicle_preferences.py create mode 100644 openpilot/starpilot/vehicle_startup.py diff --git a/opendbc_repo/opendbc/car/gm/tests/test_gm_auto_identity.py b/opendbc_repo/opendbc/car/gm/tests/test_gm_auto_identity.py new file mode 100644 index 0000000000..d4275e6015 --- /dev/null +++ b/opendbc_repo/opendbc/car/gm/tests/test_gm_auto_identity.py @@ -0,0 +1,174 @@ +import hashlib +import json +import unittest + +from opendbc.car.gm.tests import test_bolt_auto_fingerprint_pure as fingerprint_fixture + + +EXPECTED = { + 'gm.CHEVROLET_BOLT_CC_2017': '84a0177d0dfb1e3710a4e7b8abdb8fbbc4407c5596c1a14f10071c387279c03d', + 'gm.CHEVROLET_EQUINOX_CC': 'c9f20c9e63f76d1270d315775088e68b6b11f0472f18283164e887fe328cd0f7', + 'gm.CHEVROLET_SUBURBAN_CC': '7ce113446702764109bdad71e722479017d2d1fbf9bd73d68d540db9404d0e37', + 'gm.CADILLAC_CT6_CC': 'b88f1c68422e1f230c22f859614d66ccb789a27c4e33ce2bfd5651481fb27184', + 'gm.CHEVROLET_TRAILBLAZER_CC': '3e69cd3f84de888c97fc9bf24591c01066de2d2ea3bd872dede9be5866e902ac', + 'gm.CHEVROLET_MALIBU_CC': '652c37bd532de3fddd97bd062f349f70f960f8155912ef2a8f99708b3fc1449c', + 'gm.CADILLAC_XT5_CC': 'c642e13066242213ab13638da14cfe497ff43f9178eb44a91b7c95798bc92776', +} + + +class TestGMAutoIdentity(unittest.TestCase): + setUp = fingerprint_fixture.TestBoltAutoFingerprint.setUp + candidates = fingerprint_fixture.TestBoltAutoFingerprint.candidates + + def test_restored_tables_and_joint_discovery(self): + for identity, digest in EXPECTED.items(): + variants = self.inventory[identity] + if identity == 'gm.CHEVROLET_MALIBU_CC': + self.assertEqual(len(variants), 2) + self.assertEqual(hashlib.sha256(json.dumps(variants[1:], sort_keys=True, separators=(',', ':')).encode()).hexdigest(), + '1cfbba5e0c98945d6bbc6a9d40df14b26e9782bc256889438c91b71d54a65c9b') + original = variants[:1] if identity == 'gm.CHEVROLET_MALIBU_CC' else variants + self.assertEqual(hashlib.sha256(json.dumps(original, sort_keys=True, separators=(',', ':')).encode()).hexdigest(), digest) + for index, variant in enumerate(variants): + with self.subTest(identity=identity, variant=index): + self.assertEqual(self.candidates(variant.items()), [identity]) + + def test_existing_unique_identities_do_not_gain_collisions(self): + from opendbc.car.gm.tests.test_bolt_auto_fingerprint_pure import production_eliminator + from types import SimpleNamespace + baseline = {key: variants for key, variants in self.inventory.items() if key not in EXPECTED} + eliminate = production_eliminator(baseline) + for identity, variants in baseline.items(): + for index, variant in enumerate(variants): + remaining = list(baseline) + for address, length in variant.items(): + remaining = eliminate(SimpleNamespace(address=address, dat=bytes(length)), remaining) + if remaining == [identity]: + with self.subTest(identity=identity, variant=index): + expected = (['gm.CHEVROLET_MALIBU_HYBRID_CC', 'gm.CHEVROLET_MALIBU_CC'] + if identity == 'gm.CHEVROLET_MALIBU_HYBRID_CC' else [identity]) + self.assertEqual(self.candidates(variant.items()), expected) + + def test_pedal_frame_preserves_original_supported_tables(self): + identity = 'gm.CHEVROLET_BOLT_CC_2017' + for index, variant in enumerate(self.inventory[identity]): + with self.subTest(variant=index): + self.assertEqual(variant.get(0x201, 6), 6) + self.assertEqual(self.candidates([*variant.items(), (0x201, 6)]), [identity]) + + def test_discovered_gateway_identity_has_healthy_whole_parser(self): + from opendbc.can import CANPacker + from opendbc.car import Bus, gen_empty_fingerprint + from opendbc.car.gm.fingerprints import FINGERPRINTS + from opendbc.car.gm.interface import CarInterface + from opendbc.car.gm.values import (CAR, DBC, GMFlags, ORDINARY_CC_WORD, MALIBU_CC_F1_WORD, + MALIBU_CC_F1_STOCK_WORD, is_ordinary_cc_profile, control_flags) + from opendbc.car import structs + from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences + from opendbc.car.gm.tests.test_cc_gateway_stock import pt_frames + for identity in EXPECTED: + if identity == 'gm.CHEVROLET_BOLT_CC_2017': + continue + car = getattr(CAR, identity.removeprefix('gm.')) + for variant in FINGERPRINTS[car]: + discovered = self.candidates(variant.items()) + self.assertEqual(discovered, [identity]) + fp = gen_empty_fingerprint() + fp[0].update(variant) + f1 = car == CAR.CHEVROLET_MALIBU_CC and 0xBE not in variant + expected_flags = int(GMFlags.CC_LONG | (GMFlags.NO_ACCELERATOR_POS_MSG if f1 else 0)) + expected_word = MALIBU_CC_F1_WORD if f1 else ORDINARY_CC_WORD + for alpha, release in ((False, False), (True, False), (True, True)): + cp = CarInterface.get_params(car, fp, [], alpha, release, False) + self.assertEqual(cp.carFingerprint, car) + self.assertTrue(is_ordinary_cc_profile(cp), identity) + self.assertTrue(cp.openpilotLongitudinalControl) + self.assertFalse(cp.pcmCruise) + self.assertFalse(cp.alphaLongitudinalAvailable) + self.assertFalse(cp.dashcamOnly) + self.assertEqual(control_flags(cp), expected_flags) + self.assertEqual(cp.safetyConfigs[0].safetyModel, structs.CarParams.SafetyModel.gm) + self.assertEqual(cp.safetyConfigs[0].safetyParam, expected_word) + disabled = cp.as_reader().as_builder() + preferences = VehicleStartupPreferences(disable_bolt_long=True) + preferences.prepare(disabled, fingerprints=fp) + preferences.finalize(disabled) + self.assertTrue(is_ordinary_cc_profile(disabled), identity) + self.assertFalse(disabled.openpilotLongitudinalControl) + self.assertEqual(disabled.pcmCruise, f1) + self.assertEqual(disabled.safetyConfigs[0].safetyParam, MALIBU_CC_F1_STOCK_WORD if f1 else expected_word) + ci = CarInterface(cp) + packer = CANPacker(DBC[car][Bus.pt]) + for tick in range(40): + frames = pt_frames(packer, counter=tick % 4) + accelerator_address = packer.make_can_msg('ECMAcceleratorPos', 0, {})[0] + if accelerator_address not in variant: + frames = [frame for frame in frames if frame[0] != accelerator_address] + if 0xF1 in variant: + frames.append(packer.make_can_msg('EBCMBrakePedalPosition', 0, {})) + if cp.flags & GMFlags.HAS_BSM: + bsm = packer.make_can_msg('BCMBlindSpotMonitor', 0, {}) + observed_length = variant[bsm[0]] + frames.append((bsm[0], bsm[1][:observed_length], bsm[2])) + out = ci.update([(1_000_000_000 + tick * 10_000_000, frames)]) + self.assertTrue(out.canValid, identity) + self.assertFalse(out.canTimeout, identity) + + def test_discovered_2017_bolt_parser_with_actual_pedal_and_camera_configuration(self): + from opendbc.car import Bus, gen_empty_fingerprint + from opendbc.car.gm.fingerprints import FINGERPRINTS + from opendbc.car.gm.interface import CarInterface + from opendbc.car.gm.tests import test_bolt_pedal as pedal_fixture + from opendbc.car.gm.values import CAR, GMFlags + from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences + + car = CAR.CHEVROLET_BOLT_CC_2017 + fixture = pedal_fixture.TestBoltPedalStartupParser() + for variant_index, variant in enumerate(FINGERPRINTS[car]): + self.assertEqual(self.candidates(variant.items()), ['gm.' + car.name]) + present = 0x201 in variant + for saved in (False, True): + for alpha in (False, True): + for camera in ((True,) if present else (False, True)): + for disabled in (False, True): + with self.subTest(variant=variant_index, pedal=saved, alpha=alpha, camera=camera, disabled=disabled): + fp = gen_empty_fingerprint() + fp[0].update(variant) + if camera: + fp[2].update({0x180: 4, 0x320: 8}) + cp = CarInterface.get_params(car, fp, [], alpha, False, False) + VehicleStartupPreferences(disable_bolt_long=disabled).prepare(cp, fingerprints=fp) + self.assertEqual(bool(cp.flags & GMFlags.PEDAL_LONG), present) + self.assertTrue(cp.flags & GMFlags.HAS_BSM) + ci, state = fixture.stream(cp, blindspot=True, pedal_present=present, camera_present=camera) + self.assertTrue(state.canValid, (variant_index, saved, alpha, camera, disabled, cp.to_dict())) + self.assertFalse(state.canTimeout) + self.assertEqual(ci.CS.pedal_sensor_healthy, present) + self.assertNotIn('ASCMActiveCruiseControlStatus', ci.can_parsers[Bus.cam].vl) + _, missing_bsm = fixture.stream(cp, missing=(0, 'BCMBlindSpotMonitor'), blindspot=True, + pedal_present=present, camera_present=camera) + self.assertFalse(missing_bsm.canValid) + + def test_sparse_hybrid_fingerprint_cannot_select_an_automatic_identity(self): + from types import SimpleNamespace + from opendbc.car.car_helpers import can_fingerprint + identity = 'gm.CHEVROLET_MALIBU_HYBRID_CC' + variants = self.inventory[identity] + self.assertEqual(len(variants), 1) + messages = [SimpleNamespace(address=address, dat=bytes(length), src=0) + for address, length in variants[0].items()] + self.assertEqual(self.candidates(variants[0].items()), + [identity, 'gm.CHEVROLET_MALIBU_CC']) + selected, observed = can_fingerprint(lambda **_kwargs: [messages]) + self.assertIsNone(selected) + self.assertEqual(observed[0], variants[0]) + + def test_ambiguous_and_manual_identities_remain_unadvertised(self): + for name in ('CHEVROLET_BOLT_ACC_2022_2023', 'CHEVROLET_BOLT_ACC_2022_2023_PEDAL', + 'CHEVROLET_BOLT_CC_2022_2023', 'CHEVROLET_TRAX', + 'CHEVROLET_VOLT_CC'): + self.assertNotIn('gm.' + name, self.inventory) + + +if __name__ == '__main__': + unittest.main() diff --git a/opendbc_repo/opendbc/car/hyundai/tests/test_blended_longitudinal_lifecycle.py b/opendbc_repo/opendbc/car/hyundai/tests/test_blended_longitudinal_lifecycle.py new file mode 100644 index 0000000000..bc216d5530 --- /dev/null +++ b/opendbc_repo/opendbc/car/hyundai/tests/test_blended_longitudinal_lifecycle.py @@ -0,0 +1,331 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from opendbc.car.hyundai.blended_longitudinal import BlendedLongitudinalOwner, DISABLE, ENABLE, Phase, Outcome +from opendbc.car.hyundai.values import HyundaiFlags + + +class TestBlendedLifecycle(unittest.TestCase): + def setUp(self): + self.scope = patch('opendbc.car.hyundai.blended_longitudinal.is_blended', return_value=True) + self.scope.start() + self.addCleanup(self.scope.stop) + self.cp = SimpleNamespace(flags=HyundaiFlags.CANFD_LKA_STEER_MSG.value, + openpilotLongitudinalControl=True, pcmCruise=False, + safetyConfigs=[SimpleNamespace(safetyParam=0x2014)]) + self.calls = [] + + def owner(self, exchange): + return BlendedLongitudinalOwner(self.cp, SimpleNamespace(ECAN=1), exchange, + restore_exchange=lambda *a, **kw: exchange(*a, **kw, com_cont_req=ENABLE, reset=True, require_response=True)) + + def test_rejected_start_restores_stock_and_exact_ecu(self): + def exchange(*args, **kwargs): + self.calls.append(kwargs) + return kwargs['com_cont_req'] == ENABLE + owner = self.owner(exchange) + self.assertEqual(owner.begin(None, None, unpublished=True, admission=True).outcome, Outcome.STOCK) + self.assertEqual([c['com_cont_req'] for c in self.calls], [DISABLE, ENABLE]) + self.assertTrue(self.calls[-1]['require_response']) + self.assertEqual(ENABLE, b'\x28\x00\x01') + self.assertTrue(all(c['bus'] == 1 and c['addr'] == 0x730 and c['reset'] is True for c in self.calls)) + self.assertEqual(self.cp.safetyConfigs[0].safetyParam, 0x2010) + self.assertFalse(self.cp.openpilotLongitudinalControl) + self.assertTrue(self.cp.pcmCruise) + self.assertFalse(owner.active()) + self.assertEqual(owner.phase, Phase.RESTORED) + + def test_cancel_during_disable_never_grants_long_ownership(self): + def exchange(*args, **kwargs): + self.calls.append(kwargs) + if kwargs['com_cont_req'] == DISABLE: + owner.cancel() + return True + owner = self.owner(exchange) + self.assertEqual(owner.begin(None, None, unpublished=True, admission=True).outcome, Outcome.STOCK) + self.assertEqual(len(self.calls), 2) + self.assertFalse(owner.active()) + self.assertTrue(owner.restore_confirmed) + + def test_success_then_deinit_and_restore_failure_not_reported_success(self): + owner = self.owner(lambda *args, **kw: kw['com_cont_req'] == DISABLE) + self.assertEqual(owner.begin(None, None, unpublished=True, admission=True).outcome, Outcome.OWNED) + self.assertTrue(owner.active()) + self.assertFalse(owner.restore(None, None)) + self.assertEqual(owner.phase, Phase.FAILED) + self.assertFalse(owner.active()) + + def test_published_cp_or_missing_admission_cannot_start(self): + owner = self.owner(lambda *a, **k: self.fail('Unexpected ECU exchange')) + with self.assertRaises(RuntimeError): + owner.begin(None, None, unpublished=False, admission=True) + self.assertEqual(self.cp.safetyConfigs[0].safetyParam, 0x2014) + self.assertEqual(owner.begin(None, None, unpublished=True, admission=False).outcome, Outcome.STOCK_UNTOUCHED) + self.assertEqual(self.cp.safetyConfigs[0].safetyParam, 0x2010) + + def test_partial_disable_and_restore_timeout_cannot_admit_stock_cp(self): + owner = self.owner(lambda *args, **kwargs: False) + result = owner.begin(None, None, unpublished=True, admission=True) + self.assertEqual(result.outcome, Outcome.ABORT_UNCERTAIN) + self.assertFalse(result.admissible) + with self.assertRaises(TypeError): + bool(result) + with self.assertRaises(RuntimeError): + owner.publication_cp() + self.assertFalse(owner.active()) + + def test_cancelled_constructor_with_failed_restore_forbids_continuation(self): + def exchange(*args, **kwargs): + if kwargs['com_cont_req'] == DISABLE: + owner.cancel() + return True + return False + owner = self.owner(exchange) + result = owner.begin(None, None, unpublished=True, admission=True) + self.assertEqual(result.outcome, Outcome.ABORT_UNCERTAIN) + with self.assertRaises(RuntimeError): + owner.publication_cp() + + def test_real_protocol_empty_response_cannot_confirm_restoration(self): + from opendbc.car.hyundai.blended_disable_ecu import disable_ecu + class Query: + def __init__(self, *args, **kwargs): + self.request = args[4][0] + def get_data(self, *args, **kwargs): + return {(0x738, None): b''} if self.request == b'\x10\x03' else {} + with patch('opendbc.car.hyundai.blended_disable_ecu.IsoTpParallelQuery', Query), \ + patch('opendbc.car.hyundai.blended_disable_ecu.time.sleep'): + self.assertFalse(disable_ecu(None, None, bus=1, addr=0x730, com_cont_req=ENABLE, + reset=True, retry=1, require_response=True)) + + def test_restore_query_requires_unsuppressed_request_and_exact_ack_prefix(self): + from opendbc.car.hyundai.blended_disable_ecu import restore_ecu + class Query: + reply = True + def __init__(self, *args, **kwargs): + self.request, self.response = args[4][0], args[5][0] + if self.request == ENABLE and self.response != b'\x68\x00': + raise AssertionError('Restoration requires matched acknowledgment') + def get_data(self, *args, **kwargs): + return {(0x738, None): b''} if self.request == b'\x10\x03' or self.reply else {} + with patch('opendbc.car.hyundai.blended_disable_ecu.IsoTpParallelQuery', Query): + self.assertEqual(ENABLE, b'\x28\x00\x01') + self.assertTrue(restore_ecu(None, None, bus=1, addr=0x730, retry=1)) + Query.reply = False + self.assertFalse(restore_ecu(None, None, bus=1, addr=0x730, retry=1)) + + def test_real_precreate_builder_clone_has_no_source_alias(self): + from opendbc.car import gen_empty_fingerprint + from opendbc.car.hyundai.interface import CarInterface + from opendbc.car.hyundai.values import CAR + from opendbc.car.hyundai.blended_longitudinal import candidate_from_stock + cp = CarInterface.get_params(CAR.HYUNDAI_PALISADE_2023, gen_empty_fingerprint(), [], False, False, False) + before = cp.to_bytes() + self.assertIsNone(candidate_from_stock(cp, alpha_requested=True)) + candidate = candidate_from_stock(cp, alpha_requested=True, native_qualified=True) + self.assertIsNotNone(candidate) + cp.clear_write_flag() + self.assertEqual(cp.to_bytes(), before) + self.assertTrue(candidate.openpilotLongitudinalControl) + self.assertEqual(candidate.safetyConfigs[0].safetyParam, 0x2004) + candidate.pcmCruise = True + cp.clear_write_flag() + self.assertEqual(cp.to_bytes(), before) + + def test_alpha_candidate_radar_and_final_fields_preserve_stock_builder(self): + from opendbc.car.hyundai.tests.test_palisade_2023 import params + from opendbc.car.hyundai.blended_longitudinal import candidate_from_stock + for hda2 in (False, True): + for radar_unavailable in (False, True): + cp = params('hdaii' if hda2 else 'hdai') + cp.radarUnavailable = radar_unavailable + before = cp.to_dict() + candidate = candidate_from_stock(cp, alpha_requested=True, native_qualified=True) + self.assertIsNotNone(candidate) + self.assertTrue(candidate.radarUnavailable) + self.assertTrue(candidate.openpilotLongitudinalControl) + self.assertFalse(candidate.pcmCruise) + self.assertEqual(candidate.safetyConfigs[0].safetyParam, 0x2014 if hda2 else 0x2004) + expected = dict(before) + expected.update(alphaLongitudinalAvailable=True, openpilotLongitudinalControl=True, + pcmCruise=False, radarUnavailable=True, stopAccel=candidate.stopAccel, + safetyConfigs=candidate.to_dict()['safetyConfigs']) + self.assertEqual(candidate.to_dict(), expected) + self.assertEqual(cp.to_dict(), before) + + def test_actual_card_constructor_failure_closes_registered_owner(self): + import openpilot.selfdrive.car.card as card + calls = [] + class PartialCar: + def __init__(self): + self.vehicle_startup = SimpleNamespace(close=lambda: calls.append('closed')) + raise RuntimeError('Constructor failed after owner registration') + with patch.object(card, 'Car', PartialCar), patch.object(card, 'config_realtime_process'): + with self.assertRaisesRegex(RuntimeError, 'Constructor failed'): + card.main() + self.assertEqual(calls, ['closed']) + + def test_actual_prepare_without_admission_configure_close_leaves_stock_untouched(self): + from opendbc.car import gen_empty_fingerprint + from opendbc.car.hyundai.interface import CarInterface + from opendbc.car.hyundai.values import CAR + from opendbc.car.hyundai.blended_longitudinal import BlendedStartup + cp = CarInterface.get_params(CAR.HYUNDAI_PALISADE_2023, gen_empty_fingerprint(), [], False, False, False) + startup = BlendedStartup(cp, cp, (lambda *a: self.fail('Unexpected receive'), + lambda *a: self.fail('Unexpected send'))) + selected = startup.prepare(admission=lambda: False) + self.assertIs(selected, cp) + self.assertEqual(startup.owner.result.outcome, Outcome.STOCK_UNTOUCHED) + self.assertFalse(startup.owner.takeover_attempted) + self.assertFalse(startup.owner.restore_confirmed) + startup.configure(SimpleNamespace(CC=SimpleNamespace())) + startup.close() + startup.close() + self.assertFalse(cp.openpilotLongitudinalControl) + + def test_post_handoff_close_without_diagnostic_admission_never_sends_uds(self): + from opendbc.car import gen_empty_fingerprint + from opendbc.car.hyundai.interface import CarInterface + from opendbc.car.hyundai.values import CAR + from opendbc.car.hyundai.blended_longitudinal import BlendedStartup + cp = CarInterface.get_params(CAR.HYUNDAI_PALISADE_2023, gen_empty_fingerprint(), [], False, False, False) + startup = BlendedStartup(cp, cp, (lambda *a: self.fail('Unexpected receive'), + lambda *a: self.fail('Unexpected send'))) + startup.owner.takeover_attempted = True + startup.owner.phase = Phase.OWNED + startup.diagnostic_admission = lambda: False + with self.assertRaisesRegex(RuntimeError, 'Diagnostic admission unavailable'): + startup.close() + self.assertEqual(startup.owner.result.outcome, Outcome.ABORT_UNCERTAIN) + + def test_actual_parser_rewarm_requires_every_source_after_transaction_floor(self): + from opendbc.car.hyundai.tests.test_palisade_2023 import params, feed + from opendbc.car.hyundai.interface import CarInterface + from opendbc.car.hyundai.blended_longitudinal import BlendedStartup, candidate_from_stock, TakeoverResult + cp = params('hdai') + ci = CarInterface(cp) + for tick in range(12): + feed(ci, tick) + candidate = candidate_from_stock(cp, alpha_requested=True, native_qualified=True) + startup = BlendedStartup(cp, candidate, (lambda *a: None, lambda *a: None)) + startup.owner.phase = Phase.OWNED + startup.owner.result = TakeoverResult(Outcome.OWNED) + startup.source_floor_ns = 1_100_000_000 + self.assertTrue(startup.sources_current(ci, 1_120_000_000)) + startup.source_floor_ns = 1_110_000_000 + self.assertFalse(startup.sources_current(ci, 1_120_000_000)) + + def test_actual_publish_ack_occurs_after_send_lock_release_before_worker_stop(self): + import openpilot.selfdrive.car.card as card + from openpilot.starpilot.vehicle_startup import VehicleStartupOwner + context = VehicleStartupOwner() + calls = [] + def sent(frames, *, valid): + acquired = context.send_lock.acquire(blocking=False) + self.assertTrue(acquired, 'Worker join would deadlock while sender lock is held') + if acquired: + context.send_lock.release() + self.assertEqual(calls, ['published']) + calls.append('ack') + context.owner = SimpleNamespace(sent=sent) + car = card.Car.__new__(card.Car) + car.vehicle_startup = context + car.pm = SimpleNamespace(send=lambda *args: calls.append('published')) + card.Car.publish_sendcan(car, [], valid=True) + self.assertEqual(calls, ['published', 'ack']) + + def test_handoff_waits_for_sent_tester_and_later_panda_loss_inhibits(self): + from opendbc.car.hyundai.tests.test_palisade_2023 import params + from opendbc.car.hyundai.blended_longitudinal import BlendedStartup, candidate_from_stock, TakeoverResult + cp = params('hdai') + candidate = candidate_from_stock(cp, alpha_requested=True, native_qualified=True) + startup = BlendedStartup(cp, candidate, (lambda *a: None, lambda *a: None)) + startup.owner.phase = Phase.OWNED + startup.owner.result = TakeoverResult(Outcome.OWNED) + self.assertTrue(startup.before_control(configured=True, sources_current=True, control_current=True)) + self.assertFalse(startup.handed_off) + self.assertFalse(startup.stop_event.is_set()) + startup.sent([], valid=True) + self.assertFalse(startup.handed_off) + from opendbc.car import make_tester_present_msg + tester = make_tester_present_msg(startup.owner.address, startup.owner.bus, suppress_response=True) + startup.sent([tester], valid=False) + self.assertFalse(startup.handed_off) + startup.sent([tester], valid=True) + self.assertTrue(startup.handed_off) + self.assertTrue(startup.stop_event.is_set()) + self.assertFalse(startup.before_control(configured=False, sources_current=True, control_current=True)) + startup.check() + self.assertTrue(startup.owner.active()) + self.assertTrue(startup.before_control(configured=True, sources_current=True, control_current=True)) + + def test_transient_sources_pause_output_and_keep_only_owned_diagnostics(self): + from opendbc.car.hyundai.tests.test_palisade_2023 import params + from opendbc.car.hyundai.blended_longitudinal import BlendedStartup, candidate_from_stock, TakeoverResult + from opendbc.car import make_tester_present_msg + cp = params('hdai') + candidate = candidate_from_stock(cp, alpha_requested=True, native_qualified=True) + sent = [] + now = [10.] + startup = BlendedStartup(cp, candidate, (lambda *a: None, lambda frames: sent.extend(frames))) + startup.clock = lambda: now[0] + startup.started_at = 1. + startup.owner.phase = Phase.OWNED + startup.owner.result = TakeoverResult(Outcome.OWNED) + tester = make_tester_present_msg(startup.owner.address, startup.owner.bus, suppress_response=True) + startup.sent([tester], valid=True) + self.assertTrue(startup.handed_off) + for missing in ('sources_current', 'control_current'): + evidence = {'configured': True, 'sources_current': True, 'control_current': True} + evidence[missing] = False + self.assertFalse(startup.before_control(**evidence)) + startup.check() + self.assertTrue(startup.owner.active()) + now[0] = 10.9 + startup.maintain(configured=True) + self.assertEqual(sent, []) + now[0] = 11. + startup.maintain(configured=True) + self.assertEqual(sent, [tester]) + startup.maintain(configured=True) + self.assertEqual(sent, [tester]) + now[0] = 11.1 + startup.maintain(configured=False) + self.assertEqual(sent, [tester]) + self.assertFalse(startup.before_control(configured=False, sources_current=True, control_current=True)) + # Continuous diagnostic cadence is maintained during a long control/input + # pause. This does not assert session continuity across a long Panda outage. + for second in range(12, 81): + now[0] = float(second) + startup.maintain(configured=True) + self.assertEqual(sent, [tester] * 70) + self.assertIsNone(startup.fault, 'Startup-only deadline must not end a running drive') + self.assertTrue(startup.before_control(configured=True, sources_current=True, control_current=True)) + now[0] = 80.5 + startup.sent([tester], valid=True) + now[0] = 81.4 + startup.maintain(configured=True) + self.assertEqual(sent, [tester] * 70, 'A normal sent tester must update diagnostic cadence') + + def test_actual_card_maintenance_runs_before_initialization_and_never_applies(self): + import openpilot.selfdrive.car.card as card + calls = [] + CS = SimpleNamespace() + car = card.Car.__new__(card.Car) + car.state_update = lambda: (CS, None) + car.state_publish = lambda *args: calls.append('state') + car.CP = SimpleNamespace(passive=False) + class SM(dict): + seen = {'onroadEvents': False} + car.sm = SM(onroadEvents=[]) + car.vehicle_startup = SimpleNamespace(owner=object(), maintain=lambda **kw: calls.append(('maintain', kw))) + car.startup_panda_configured = lambda: True + car.controls_update = lambda *args: self.fail('Controls are not initialized') + card.Car.step(car) + self.assertEqual(calls, [('maintain', {'configured': True}), 'state']) + self.assertIs(car.CS_prev, CS) + + +if __name__ == '__main__': + unittest.main() diff --git a/opendbc_repo/opendbc/car/hyundai/tests/test_g90_startup.py b/opendbc_repo/opendbc/car/hyundai/tests/test_g90_startup.py new file mode 100644 index 0000000000..c11fdf252e --- /dev/null +++ b/opendbc_repo/opendbc/car/hyundai/tests/test_g90_startup.py @@ -0,0 +1,190 @@ +"""Actual get_car/owner/CI publication path with explicit diagnostic transport replies.""" +import ast +import inspect +import textwrap +import time +import unittest +from unittest.mock import patch + +from opendbc.can import CANPacker +from opendbc.car import Bus, gen_empty_fingerprint, structs +from opendbc.car import car_helpers, disable_ecu as disable_module +from opendbc.car.can_definitions import CanData +from opendbc.car.hyundai import g90_startup +from opendbc.car.hyundai.g90_startup import Outcome +from opendbc.car.hyundai.ioniq6_handoff import TimestampedCanPacket +from opendbc.car.hyundai.interface import CarInterface +from opendbc.car.hyundai.tests.test_g90_longitudinal import params +from opendbc.car.hyundai.values import CAR, DBC, HyundaiSafetyFlags +from openpilot.selfdrive.car.card import Car +from openpilot.starpilot.vehicle_startup import VehicleStartupOwner + + +class Replies: + def __init__(self, mode): + self.mode = mode + self.requests = [] + + def query(self, send, recv, bus, addresses, requests, responses, **kwargs): + request = requests[0] + expected = {b'\x10\x03': b'\x50\x03', b'\x28\x83\x01': b'', b'\x28\x00\x01': b'\x68\x00'} + assert responses == [expected[request]], (request, responses) + self.requests.append(request) + owner = self + class Query: + def get_data(self, timeout): + send([CanData(addresses[0][0], bytes((len(request),)) + request + bytes(7-len(request)), bus)]) + if request == b'\x10\x03': + return {} if owner.mode == 'untouched' else {(0x7d0, None): b''} + if request == b'\x28\x83\x01': + if owner.mode != 'sent': + raise OSError('Explicit simulated failure after communication request send') + return {} + if request == b'\x28\x00\x01': + return {(0x7d0, None): b''} if owner.mode == 'restored' else {} + raise AssertionError(request) + return Query() + + +class TestG90Startup(unittest.TestCase): + # Mac has no BOOTTIME constant. This fixture-only shim uses its monotonic + # clock for both producer and consumer; Linux exercises native BOOTTIME. + @patch.object(time, 'CLOCK_BOOTTIME', getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC), create=True) + def run_prepublication(self, mode, *, warm_mode=None, admission=lambda: True, requested=True): + replies = Replies(mode) + holder = VehicleStartupOwner() + holder_ci = [] + original = params() + sent = [] + packer = CANPacker(DBC[CAR.GENESIS_G90][Bus.pt]) + def recv(wait_for_one=False): + if not holder_ci: + return [] + ci = holder_ci[0] + frames = [] + for parser in ci.can_parsers.values(): + for address in parser.addresses: + message = parser.dbc.addr_to_msg[address] + packet = CanData(*packer.make_can_msg(message.name, parser.bus, {})) + if warm_mode == 'wrongbus' and message.name in ('SCC11', 'SCC12'): + packet = CanData(packet.address, packet.dat, 1) + frames.extend([packet] * (7 if warm_mode == 'counter' else 1)) + stamp = 1 if warm_mode == 'stale' else time.clock_gettime_ns(time.CLOCK_BOOTTIME) + return [TimestampedCanPacket(frames, stamp)] + def hook(cp, candidate, fingerprints, firmware): + return holder.prepare(cp, CarInterface, (recv, sent.extend), requested=requested, admission=admission) + fingerprint = (CAR.GENESIS_G90, gen_empty_fingerprint(), '0'*17, [], original.fingerprintSource, True) + with patch.object(car_helpers, 'fingerprint', return_value=fingerprint), \ + patch.object(disable_module, 'IsoTpParallelQuery', side_effect=replies.query), \ + patch.object(g90_startup, 'IsoTpParallelQuery', side_effect=replies.query): + ci = car_helpers.get_car(recv, sent.extend, lambda *args: None, True, False, pre_create_hook=hook) + holder_ci.append(ci) + if not requested: + # Actual Card finalizes passive before holder.configure; no native output. + ci.CP.passive = True + ci.CP.safetyConfigs[0].safetyModel = structs.CarParams.SafetyModel.noOutput + ci.CP.safetyConfigs[0].safetyParam = 0 + if warm_mode: + with patch.object(g90_startup.time, 'monotonic', side_effect=[0., 0., 4.]): + holder.configure(ci) + else: + holder.configure(ci) + holder.seal_publication() + return ci, holder, replies, original + + def test_actual_prepublication_sent_success_preserves_cp_and_protocol(self): + ci, holder, replies, original = self.run_prepublication('sent') + self.assertIs(holder.owner.outcome, Outcome.SENT_UNCONFIRMED) + self.assertTrue(ci.CP.openpilotLongitudinalControl) + self.assertFalse(ci.CP.pcmCruise) + self.assertIn(b'\x28\x83\x01', replies.requests) + self.assertNotIn(b'\x28\x03\x01', replies.requests) + self.assertEqual(ci.CP.safetyConfigs[0].safetyParam, original.safetyConfigs[0].safetyParam) + + def test_actual_failed_prepublication_restores_then_publishes_stock(self): + for mode, outcome in (('untouched', Outcome.STOCK_UNTOUCHED), ('restored', Outcome.STOCK_RESTORED)): + ci, holder, replies, original = self.run_prepublication(mode) + self.assertIs(holder.owner.outcome, outcome) + self.assertFalse(ci.CP.openpilotLongitudinalControl) + self.assertTrue(ci.CP.pcmCruise) + self.assertFalse(ci.CP.safetyConfigs[0].safetyParam & HyundaiSafetyFlags.LONG.value) + self.assertTrue(original.openpilotLongitudinalControl) # Caller clone not mutated. + self.assertTrue(holder.owner.ready) + self.assertEqual(b'\x28\x00\x01' in replies.requests, mode == 'restored') + + def test_ambiguous_failure_aborts_before_constructor_and_explicit_ci_requires_owner(self): + with self.assertRaisesRegex(RuntimeError, 'restoration unverified'): + self.run_prepublication('uncertain') + ci = CarInterface(params()) + with self.assertRaisesRegex(RuntimeError, 'matching prepared'): + VehicleStartupOwner().configure(ci) + holder = VehicleStartupOwner() + stock = params(alpha=False) + holder.configure(CarInterface(stock)) # Ordinary stock and all siblings need no owner. + + def test_stale_wrongbus_and_counter_loss_cannot_publish_stock(self): + for warm_mode in ('stale', 'wrongbus', 'counter'): + with self.subTest(warm_mode=warm_mode), self.assertRaisesRegex(RuntimeError, 'stock SCC'): + self.run_prepublication('restored', warm_mode=warm_mode) + + def test_revoked_diagnostic_admission_cannot_restore_partial_failure(self): + admitted = iter((True, False)) + with self.assertRaisesRegex(RuntimeError, 'restoration unverified'): + self.run_prepublication('restored', admission=lambda: next(admitted)) + + def test_explicit_prepared_owner_matches_exact_cp(self): + ci, holder, replies, original = self.run_prepublication('sent') + VehicleStartupOwner(holder.owner).configure(ci) + other = params() + other.safetyConfigs[0].safetyParam ^= 1 + with self.assertRaisesRegex(RuntimeError, 'matching prepared'): + VehicleStartupOwner(holder.owner).configure(CarInterface(other)) + + def test_same_object_mutation_cannot_rebind_prepared_decision(self): + for mutation in ('safety', 'ownership'): + ci, holder, replies, original = self.run_prepublication('sent') + self.assertIs(ci.CP, holder.owner.cp) + if mutation == 'safety': + ci.CP.safetyConfigs[0].safetyParam ^= 1 + else: + ci.CP.openpilotLongitudinalControl = False + ci.CP.pcmCruise = True + self.assertFalse(holder.owner.prepared_for(ci.CP)) + with self.assertRaisesRegex(RuntimeError, 'prepared'): + holder.seal_publication() + with self.assertRaisesRegex(RuntimeError, 'prepared'): + holder.configure(ci) + + def test_distinct_equal_cp_mutation_cannot_pass_publication(self): + for mutation in ('safety', 'ownership'): + ci, holder, replies, original = self.run_prepublication('sent') + with structs.CarParams.from_bytes(ci.CP.to_bytes()) as reader: + equal_cp = reader.as_builder() + distinct_ci = CarInterface(equal_cp) + self.assertIsNot(distinct_ci.CP, holder.owner.cp) + self.assertTrue(holder.owner.prepared_for(distinct_ci.CP)) + holder.configure(distinct_ci) + if mutation == 'safety': + distinct_ci.CP.safetyConfigs[0].safetyParam ^= 1 + else: + distinct_ci.CP.openpilotLongitudinalControl = False + distinct_ci.CP.pcmCruise = True + self.assertTrue(holder.owner.prepared_for(holder.owner.cp)) + with self.assertRaisesRegex(RuntimeError, 'prepared'): + holder.seal_publication() + + def test_disabled_request_passive_finalization_never_sends_diagnostics(self): + ci, holder, replies, original = self.run_prepublication('sent', requested=False) + self.assertIsNone(holder.owner) + self.assertEqual(replies.requests, []) + self.assertTrue(ci.CP.passive) + self.assertEqual(ci.CP.safetyConfigs[0].safetyModel, structs.CarParams.SafetyModel.noOutput) + self.assertEqual(ci.CP.safetyConfigs[0].safetyParam, 0) + + def test_actual_card_publication_order_and_generic_explicit_owner_api(self): + source = inspect.getsource(Car.__init__) + self.assertIn('startup_owner=None', source) + self.assertLess(source.index('self.vehicle_startup.configure(self.CI)'), source.index('self.CP.to_bytes()')) + self.assertLess(source.index('self.vehicle_startup.seal_publication()'), source.index('self.CP.to_bytes()')) + tree = ast.parse(textwrap.dedent(source)) + self.assertFalse(any(isinstance(node, ast.Constant) and node.value == 'GENESIS_G90' for node in ast.walk(tree))) diff --git a/opendbc_repo/opendbc/safety/tests/test_hyundai_blended_cruise_ack.py b/opendbc_repo/opendbc/safety/tests/test_hyundai_blended_cruise_ack.py new file mode 100644 index 0000000000..97190df891 --- /dev/null +++ b/opendbc_repo/opendbc/safety/tests/test_hyundai_blended_cruise_ack.py @@ -0,0 +1,104 @@ +"""Actual CAN/native acknowledgment and cruise consumers; synthetic transport clocks.""" +import ast +import inspect +import textwrap +from types import SimpleNamespace +import unittest + +from opendbc.car.hyundai.values import Buttons +from opendbc.safety.tests.test_hyundai_blended_alpha_paired_ack import AcknowledgedCancelStream +from openpilot.selfdrive.car.card import Car +from openpilot.selfdrive.car.cruise import VCruiseHelper, V_CRUISE_INITIAL +from openpilot.starpilot.vehicle_startup import VehicleStartupOwner + + +def reached_card_initialize(): + # Execute the exact reached Card initialization block, including its actual + # facade call. The unrelated hardware/transport state_update loop is excluded. + tree = ast.parse(textwrap.dedent(inspect.getsource(Car.state_update))) + block = next(node for node in tree.body[0].body if isinstance(node, ast.If) and + any(isinstance(call, ast.Call) and ast.unparse(call.func) == 'self.v_cruise_helper.initialize_v_cruise' + for call in ast.walk(node))) + fn = ast.FunctionDef(name='initialize', args=ast.arguments(posonlyargs=[], args=[ast.arg(arg='self')], + kwonlyargs=[], kw_defaults=[], defaults=[]), + body=[block], decorator_list=[]) + namespace = {} + exec(compile(ast.fix_missing_locations(ast.Module(body=[fn], type_ignores=[])), '', 'exec'), namespace) + return namespace['initialize'] + + +class TestBlendedCruiseAcknowledgment(unittest.TestCase): + + def acknowledged(self, topology, button): + stream = AcknowledgedCancelStream(topology) + frames = stream.frames + def moving_frames(*args): + return [stream.packer.make_can_msg('WHL_SPD11', bus, dict.fromkeys( + ('WHL_SPD_FL', 'WHL_SPD_FR', 'WHL_SPD_RL', 'WHL_SPD_RR'), 20)) if address == 0x386 + else (address, data, bus) for address, data, bus in frames(*args)] + stream.frames = moving_frames + stream.step(button=button) + release = stream.step() + self.assertFalse(release.buttonEnable) + state = stream.wait_ack() + self.assertTrue(state.buttonEnable) + self.assertTrue(stream.safety.get_controls_allowed()) + self.assertEqual(list(state.buttonEvents), []) + return stream, state + + def invoke_card(self, stream, previous): + holder = VehicleStartupOwner() + holder.owner = stream.owner + helper = VCruiseHelper(stream.ci.CP) + helper.v_cruise_kph = helper.v_cruise_kph_last = 90. + instance = SimpleNamespace(sm={'carControl': SimpleNamespace(enabled=True)}, + CC_prev=SimpleNamespace(enabled=False), CS_prev=previous, + experimental_mode=False, vehicle_startup=holder, v_cruise_helper=helper) + reached_card_initialize()(instance) + self.assertIsNone(holder.consume_cruise_resume()) + return helper.v_cruise_kph + + def test_actual_acknowledged_buttons_preserve_speed_across_feedback_delay(self): + for topology in ('hdai', 'hdaii'): + for button in (Buttons.CANCEL, Buttons.RES_ACCEL, Buttons.SET_DECEL): + for delay in (1, 4): + with self.subTest(topology=topology, button=button, delay=delay): + stream, state = self.acknowledged(topology, button) + previous = state + for _ in range(delay): + previous = stream.step() # Actual fresh host-enabled acknowledgment must retain hint. + result = self.invoke_card(stream, previous) + self.assertEqual(result, V_CRUISE_INITIAL if button == Buttons.SET_DECEL else 90.) + + def test_acknowledged_hint_invalidation_and_single_consumption(self): + for mode in ('edge', 'pedal', 'profile', 'expiry'): + stream, state = self.acknowledged('hdai', Buttons.CANCEL) + if mode == 'edge': + stream.step(button=Buttons.CANCEL) + elif mode == 'pedal': + stream.step(brake=True) + elif mode == 'profile': + stream.step(evidence='wrong_profile') + else: + for _ in range(51): + stream.step() + self.assertIsNone(stream.owner.consume_cruise_resume(), mode) + stream, state = self.acknowledged('hdaii', Buttons.RES_ACCEL) + self.assertIs(stream.owner.consume_cruise_resume(), True) + self.assertIsNone(stream.owner.consume_cruise_resume()) + + def test_default_none_preserves_ordinary_initializer(self): + from opendbc.car import structs + from opendbc.car.hyundai.tests.test_palisade_2023 import params + cp = params('hdai') + cp.pcmCruise = False + for button, expected in ((structs.CarState.ButtonEvent.Type.accelCruise, 90.), + (structs.CarState.ButtonEvent.Type.decelCruise, V_CRUISE_INITIAL)): + state = structs.CarState() + state.vEgo = 20 / 3.6 + state.buttonEvents = [structs.CarState.ButtonEvent(type=button, pressed=False)] + for explicit_none in (False, True): + helper = VCruiseHelper(cp) + helper.v_cruise_kph = helper.v_cruise_kph_last = 90. + helper.initialize_v_cruise(state, False, **({'resume': None} if explicit_none else {})) + self.assertEqual(helper.v_cruise_kph, expected) diff --git a/openpilot/starpilot/car/tesla/tests/test_screen_intent.py b/openpilot/starpilot/car/tesla/tests/test_screen_intent.py new file mode 100644 index 0000000000..53bbc3a115 --- /dev/null +++ b/openpilot/starpilot/car/tesla/tests/test_screen_intent.py @@ -0,0 +1,115 @@ +from dataclasses import replace +import pytest +from opendbc.car import structs +from opendbc.car.tesla.screen_button import apply_screen_button +from opendbc.car.tesla.tests.test_screen_button import screen_params +from openpilot.starpilot.aol.intent import AolSettings, ButtonType +from openpilot.starpilot.car.tesla.screen_intent import TeslaScreenCardIntent + + +def intent(brake=False): + cp = screen_params() + apply_screen_button(cp, True, brake) + settings = AolSettings(True, 0, 9, 0, (0, 0, 0), (0, 0, 0)) + return TeslaScreenCardIntent(cp, settings) + + +def state(*, available=True, enabled=False, press=None, brake=False): + cs = structs.CarState(canValid=True, gearShifter=structs.CarState.GearShifter.drive, brakePressed=brake) + cs.cruiseState.available = available + cs.cruiseState.enabled = enabled + if press is not None: + cs.buttonEvents = [structs.CarState.ButtonEvent(type=ButtonType.lkas, pressed=press)] + return cs + + +def test_cruise_baseline_screen_toggle_and_normal_engagement_reset(): + owner = intent() + owner.update(state().as_reader(), now_ns=1) + assert owner.allowed_latch and owner.screen_override is None + owner.update(state(press=True).as_reader(), now_ns=2) + assert not owner.allowed_latch and owner.pause_lateral + owner.update(state(press=False).as_reader(), now_ns=3) + owner.update(state(press=True).as_reader(), now_ns=4) + assert owner.allowed_latch and not owner.pause_lateral + owner.update(state(press=True).as_reader(), now_ns=5) + owner.update(state(enabled=True).as_reader(), now_ns=6) + assert owner.allowed_latch and owner.screen_override is None + + +@pytest.mark.parametrize('field', ['doorOpen', 'steerFaultPermanent', 'steeringDisengage', 'accFaulted']) +def test_fault_cannot_be_overridden_by_screen(field): + owner = intent() + cs = state(press=True) + setattr(cs, field, True) + owner.update(cs.as_reader(), now_ns=2) + assert not owner.allowed_latch + + +@pytest.mark.parametrize('brake_disengage', [False, True]) +def test_brake_choice_frozen_and_live_settings_optout(brake_disengage): + owner = intent(brake_disengage) + owner.update(state().as_reader(), now_ns=1) + owner.update(state(brake=True).as_reader(), now_ns=2) + assert owner.allowed_latch is not brake_disengage + owner.settings = replace(owner.settings, enabled=False) + owner.update(state(press=True).as_reader(), now_ns=3) + assert not owner.allowed_latch + + +def test_expired_native_permission_and_cancel_require_new_intent(): + owner = intent() + owner.update(state(available=False).as_reader(), now_ns=1) + owner.update(state(available=False, press=True).as_reader(), now_ns=2) + assert owner.allowed_latch + owner.update(state(available=False).as_reader(), now_ns=4, native_rejection_ns=3) + assert not owner.allowed_latch + cs = state(available=False, press=True) + cs.buttonEvents = [structs.CarState.ButtonEvent(type=ButtonType.cancel, pressed=True)] + owner.update(cs.as_reader(), now_ns=5) + assert not owner.allowed_latch + + +def test_actual_startup_preferences_preserve_off_and_freeze_brake(tmp_path): + from openpilot.common.params import Params + from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences + from openpilot.starpilot.car.tesla.aol import policy_for + params = Params(str(tmp_path / 'params')) + params.put_bool('AlwaysOnLateral', True, block=True) + params.put_bool('TeslaAOLScreenTap', False, block=True) + cp = screen_params() + saved = VehicleStartupPreferences.read(params, enabled=True) + saved.prepare(cp) + assert cp.safetyConfigs[0].safetyParam == 0 + assert policy_for(cp).settings_supported and not policy_for(cp).runtime_supported + params.put_bool('TeslaAOLScreenTap', True, block=True) + params.put_bool('TeslaAOLDisengageOnBrake', True, block=True) + saved = VehicleStartupPreferences.read(params, enabled=True) + cp = screen_params() + saved.prepare(cp) + assert cp.safetyConfigs[0].safetyParam == 1536 + params.put_bool('TeslaAOLDisengageOnBrake', False, block=True) + saved.finalize(cp) + assert cp.safetyConfigs[0].safetyParam == 1536 + params.put_bool('AlwaysOnLateral', False, block=True) + refreshed = VehicleStartupPreferences.read(params, enabled=True) + refreshed.prepare(cp) + assert cp.safetyConfigs[0].safetyParam == 0 + assert not cp.flags & 16 + assert policy_for(cp).settings_supported and not policy_for(cp).runtime_supported + + +def test_unbuckled_screen_gesture_preserves_independent_lateral_intent(): + owner = intent() + cs = state(available=False, press=False) + cs.seatbeltUnlatched = True + owner.update(cs.as_reader(), now_ns=1) + cs.buttonEvents[0].pressed = True + owner.update(cs.as_reader(), now_ns=2) + assert owner.allowed_latch + cs.buttonEvents = [] + owner.update(cs.as_reader(), now_ns=3) + assert owner.allowed_latch + cs.seatbeltUnlatched = False + owner.update(cs.as_reader(), now_ns=4) + assert owner.allowed_latch and not cs.cruiseState.enabled diff --git a/openpilot/starpilot/controller_extensions.py b/openpilot/starpilot/controller_extensions.py new file mode 100644 index 0000000000..83af5fc063 --- /dev/null +++ b/openpilot/starpilot/controller_extensions.py @@ -0,0 +1,106 @@ +import math +import time + +import numpy as np + +import openpilot.cereal.messaging as messaging +from openpilot.selfdrive.modeld.constants import ModelConstants +from openpilot.common.swaglog import cloudlog + +from openpilot.starpilot.controllers.toyota_cruise import ToyotaCruisePreference, capability as toyota_cruise_capability + +from opendbc.car.ford.values import CAR, FordFlags + + +class ManualTurnInputs: + def __init__(self, params): + self.params = params + self.sm = messaging.SubMaster(["modelV2", "lateralDelay"]) + self.enabled = bool(self.params.get("FordHumanTurnDetection", return_default=True)) + self.frames = 0 + + def update(self): + self.sm.update(0) + if self.frames % 100 == 0: + self.enabled = bool(self.params.get("FordHumanTurnDetection", return_default=True)) + self.frames += 1 + state = self.sm["modelV2"].meta.laneChangeState + model_ready = self.sm.alive["modelV2"] and self.sm.valid["modelV2"] + return self.enabled, int(getattr(state, "raw", state)) in (1, 2, 3), model_ready + + + def preview_curvature(self, speed: float) -> float | None: + try: + now = time.monotonic_ns() + stamp = self.sm.logMonoTime["modelV2"] + if (not self.sm.alive["modelV2"] or not self.sm.valid["modelV2"] or + not 0 < stamp <= now <= stamp + 100_000_000 or not math.isfinite(speed) or speed < 0.1): + return None + yaw = tuple(self.sm["modelV2"].orientationRate.z) + if len(yaw) != len(ModelConstants.T_IDXS) or not all(math.isfinite(value) for value in yaw): + return None + delay = 0.2 + delay_stamp = self.sm.logMonoTime["lateralDelay"] + if (self.sm.alive["lateralDelay"] and self.sm.valid["lateralDelay"] and + 0 < delay_stamp <= now <= delay_stamp + 1_000_000_000): + measured_delay = float(self.sm["lateralDelay"].lateralDelay) + if math.isfinite(measured_delay): + delay = float(np.clip(measured_delay, 0.2, 0.4)) + return float(np.interp(delay, ModelConstants.T_IDXS, yaw)) / speed + except (AttributeError, KeyError, TypeError, ValueError, OverflowError): + return None + + def lateral_snapshot(self, speed: float): + if self.preview_curvature(speed) is None: + return None + delay = 0.2 + now = time.monotonic_ns() + stamp = self.sm.logMonoTime["lateralDelay"] + if (self.sm.alive["lateralDelay"] and self.sm.valid["lateralDelay"] and + 0 < stamp <= now <= stamp + 1_000_000_000): + value = float(self.sm["lateralDelay"].lateralDelay) + if math.isfinite(value): + delay = float(np.clip(value, 0.2, 0.4)) + return self.sm["modelV2"], ModelConstants.T_IDXS, delay, self.enabled + + +def configure_controller(CI, params): + cp = CI.CP + controller = CI.CC + if controller is not None and toyota_cruise_capability(cp) is not None: + controller.reverse_cruise_input = ToyotaCruisePreference(cp, params) + if (controller is not None and cp.brand == "ford" and cp.carFingerprint == CAR.FORD_MUSTANG_MACH_E_MK1 and + not cp.flags & FordFlags.LKA_STEERING and not cp.passive and not cp.dashcamOnly and not cp.notCar and + getattr(controller, "manual_turn", None) is not None): + controller.manual_turn_inputs = ManualTurnInputs(params) + + from opendbc.car.ford.classic_lateral import qualified as classic_qualified + if controller is not None and classic_qualified(cp) and getattr(controller, "classic_lateral", None) is not None: + controller.manual_turn_inputs = ManualTurnInputs(params) + + from opendbc.car.hyundai.g90_lead import eligible as g90_lead_eligible + if controller is not None and g90_lead_eligible(cp): + from openpilot.starpilot.longitudinal.g90_lead import G90LeadInputs + try: + controller.g90_lead_inputs = G90LeadInputs() + except OSError: + controller.g90_lead_inputs = None + cloudlog.exception('Optional G90 lead input transport unavailable') + + from opendbc.car.hyundai.gv70_camera_lead import eligible as gv70_lead_eligible + if controller is not None and gv70_lead_eligible(cp): + from openpilot.starpilot.longitudinal.gv70_lead import GV70LeadInputs + try: + controller.gv70_lead_inputs = GV70LeadInputs() + except OSError: + controller.gv70_lead_inputs = None + cloudlog.exception('Optional GV70 lead input transport unavailable') + + from opendbc.car.hyundai.ev9_longitudinal import qualified as ev9_long_qualified + if controller is not None and ev9_long_qualified(cp): + from openpilot.starpilot.longitudinal.canfd_lead import CANFDLeadInputs + try: + controller.ev9_lead_inputs = CANFDLeadInputs() + except OSError: + controller.ev9_lead_inputs = None + cloudlog.exception('Optional EV9 lead input transport unavailable') diff --git a/openpilot/starpilot/lateral/tests/test_turn_assist.py b/openpilot/starpilot/lateral/tests/test_turn_assist.py new file mode 100644 index 0000000000..f39fa9d4c7 --- /dev/null +++ b/openpilot/starpilot/lateral/tests/test_turn_assist.py @@ -0,0 +1,113 @@ +import math +from pathlib import Path +from types import SimpleNamespace +import tempfile +from unittest.mock import patch + +import pytest +from opendbc.car import structs +from opendbc.car.car_helpers import interfaces +from opendbc.car.hyundai.values import CAR +from opendbc.car.vehicle_model import VehicleModel +from openpilot.common.realtime import DT_CTRL +from openpilot.selfdrive.controls.lib.latcontrol_torque import LatControlTorque +from openpilot.starpilot.lateral.torque_extension import selected_policy +from openpilot.starpilot.lateral.controller_selection import ControllerMode, turn_assist_supported +from openpilot.starpilot.lateral import ioniq6_policy as policy +from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences + + +def controller(enabled=False, mode=None, minimum=0.): + cp = interfaces[CAR.HYUNDAI_IONIQ_6].get_non_essential_params(CAR.HYUNDAI_IONIQ_6) + cp.minSteerSpeed = minimum + lac = LatControlTorque(cp.as_reader(), interfaces[CAR.HYUNDAI_IONIQ_6](cp), DT_CTRL, + controller_mode=mode, turn_assist=enabled) + cs = structs.CarState.new_message() + cs.gearShifter = structs.CarState.GearShifter.drive + cs.vEgo = 1. + return cp, lac, cs + + +def test_default_off_preserves_other_controller_math(): + cp, off, cs = controller() + _, on, _ = controller(True) + vm = VehicleModel(cp) + params = SimpleNamespace(angleOffsetDeg=0., roll=0.) + original = policy.get_ioniq_6_low_speed_angle_assist_torque + with patch.object(policy, 'get_ioniq_6_low_speed_angle_assist_torque', wraps=original) as assist: + off_output, _, off_state = off.update(True, cs, vm, params, False, -.001, False, .1) + assist.assert_not_called() + on_output, _, on_state = on.update(True, cs, vm, params, False, -.001, False, .1) + assert assist.call_count == 1 + for field in ('p', 'i', 'd', 'f', 'error', 'desiredLateralAccel', 'desiredLateralJerk'): + assert getattr(off_state, field) == getattr(on_state, field) + assert off_output != on_output + assert math.isfinite(off_output) and abs(on_output) <= 1. + + +@pytest.mark.parametrize('field,value', [ + ('vEgo', 0.), ('vEgo', .044703), ('vEgo', -1.), ('vEgo', float('nan')), + ('vEgo', float('inf')), ('standstill', True), ('steeringPressed', True), + ('steerFaultTemporary', True), ('steerFaultPermanent', True), + ('gearShifter', structs.CarState.GearShifter.reverse), + ('gearShifter', structs.CarState.GearShifter.unknown), +]) +def test_assist_rejects_nonrolling_or_faulted_state(field, value): + _, lac, cs = controller(True) + setattr(cs, field, value) + assert not selected_policy(lac).turn_assist_active(cs) + + +def test_exact_rolling_threshold_and_vehicle_minimum(): + _, lac, cs = controller(True) + cs.vEgo = .044704 + assert selected_policy(lac).turn_assist_active(cs) + _, lac, cs = controller(True, minimum=2.) + assert not selected_policy(lac).turn_assist_active(cs) + cs.vEgo = 2. + assert selected_policy(lac).turn_assist_active(cs) + + +def test_stock_and_unsupported_interfaces_never_gain_policy(): + cp, lac, _ = controller(True, ControllerMode.STANDARD) + assert selected_policy(lac) is None + assert turn_assist_supported(cp) + cp.passive = True + assert not turn_assist_supported(cp) + cp.passive = False + cp.steerControlType = structs.CarParams.SteerControlType.angle + assert not turn_assist_supported(cp) + + +def test_startup_saved_exact_on_defaults_and_safe_mode(): + assert not VehicleStartupPreferences().turn_assist + snapshot = VehicleStartupPreferences(toyota_auto_hold=True) + assert snapshot.toyota_auto_hold and not snapshot.turn_assist + with tempfile.TemporaryDirectory() as root: + source = SimpleNamespace(get_param_path=lambda key: str(Path(root) / key)) + for raw in (None, b'0', b'1', b'', b'true', b'1\n', b'1'*20): + path = Path(root) / 'TurnAssist' + path.unlink(missing_ok=True) + if raw is not None: + path.write_bytes(raw) + snapshot = VehicleStartupPreferences.read(source, enabled=True) + assert snapshot.turn_assist == (raw == b'1') + path.write_bytes(b'0') + assert snapshot.turn_assist == (raw == b'1') + path.write_bytes(b'1') + assert not VehicleStartupPreferences.read(source, enabled=False).turn_assist + (Path(root) / 'SafeMode').write_bytes(b'1') + assert not VehicleStartupPreferences.read(source, enabled=True).turn_assist + + +def test_enabled_helper_retains_original_large_turn_and_unwind(): + _, lac, cs = controller(True) + assert selected_policy(lac).turn_assist_active(cs) + assert policy.get_ioniq_6_low_speed_angle_assist_torque(10., 0., 0., cs.vEgo) == -.39409320626365096 + assert policy.get_ioniq_6_low_speed_angle_assist_torque(0., 10., 0., cs.vEgo) == .07246171113121364 + + +def test_nonfinite_minimum_does_not_admit_assist(): + for minimum in (float('nan'), float('inf')): + _, lac, cs = controller(True, minimum=minimum) + assert not selected_policy(lac).turn_assist_active(cs) diff --git a/openpilot/starpilot/tests/test_controller_extensions.py b/openpilot/starpilot/tests/test_controller_extensions.py new file mode 100644 index 0000000000..b200fdbe66 --- /dev/null +++ b/openpilot/starpilot/tests/test_controller_extensions.py @@ -0,0 +1,135 @@ +import tempfile +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from openpilot.starpilot.controller_extensions import ManualTurnInputs, configure_controller +from openpilot.starpilot.tests import test_vehicle_preferences as preferences_tests +from opendbc.car.ford.values import CAR + + +class TestControllerExtensions(unittest.TestCase): + def test_real_saved_preference_default_and_disabled(self): + from openpilot.common.params import Params + with tempfile.TemporaryDirectory() as directory: + params = Params(directory) + self.assertTrue(params.get('FordHumanTurnDetection', return_default=True)) + params.put_bool('FordHumanTurnDetection', False, block=True) + self.assertFalse(params.get('FordHumanTurnDetection', return_default=True)) + + def test_actual_model_owner_and_cached_preference(self): + import openpilot.cereal.messaging as messaging + from openpilot.common.params import Params + message = messaging.new_message('modelV2') + message.modelV2.meta.laneChangeState = 'laneChangeStarting' + sm = SimpleNamespace(alive={'modelV2': True}, valid={'modelV2': True}, update=lambda timeout: None) + + class ModelReader: + alive = sm.alive + valid = sm.valid + + def update(self, timeout): + sm.update(timeout) + + def __getitem__(self, key): + return message.modelV2 + with patch.object(messaging, 'SubMaster', return_value=ModelReader()), patch.object(Params, 'get', return_value=True) as read: + inputs = ManualTurnInputs(Params()) + self.assertEqual(inputs.update(), (True, True, True)) + for _ in range(99): + inputs.update() + self.assertEqual(read.call_count, 8) + inputs.update() + self.assertEqual(read.call_count, 12) + sm.alive['modelV2'] = False + self.assertEqual(inputs.update(), (True, True, False)) + + def test_model_preview_freshness_delay_and_invalid_output(self): + import openpilot.cereal.messaging as messaging + from openpilot.selfdrive.modeld.constants import ModelConstants + now = 10_000_000_000 + model = messaging.new_message('modelV2') + model.modelV2.orientationRate.z = [20. * (value / 100.) for value in ModelConstants.T_IDXS] + delay = messaging.new_message('lateralDelay') + delay.lateralDelay.lateralDelay = .3 + class Reader: + alive = {'modelV2': True, 'lateralDelay': True} + valid = dict(alive) + logMonoTime = {'modelV2': now, 'lateralDelay': now} + def __getitem__(self, key): + return model.modelV2 if key == 'modelV2' else delay.lateralDelay + inputs = ManualTurnInputs.__new__(ManualTurnInputs) + vars(inputs)['sm'] = Reader() + with patch('openpilot.starpilot.controller_extensions.time.monotonic_ns', return_value=now): + curvature = inputs.preview_curvature(20.) + assert curvature is not None + self.assertAlmostEqual(curvature, .003) + inputs.sm.logMonoTime['lateralDelay'] = now - 1_000_000_001 + curvature = inputs.preview_curvature(20.) + assert curvature is not None + self.assertAlmostEqual(curvature, .002) + for stamp in (0, now + 1, now - 100_000_001): + inputs.sm.logMonoTime['modelV2'] = stamp + self.assertIsNone(inputs.preview_curvature(20.)) + inputs.sm.logMonoTime['modelV2'] = now + self.assertIsNone(inputs.preview_curvature(0.)) + model.modelV2.orientationRate.z = [float('nan')] * 33 + self.assertIsNone(inputs.preview_curvature(20.)) + + def test_assist_permission_tracks_panda_health_without_affecting_preview(self): + import openpilot.cereal.messaging as messaging + from openpilot.common.params import Params + + class Reader(dict): + def __init__(self): + super().__init__(pandaStates=[SimpleNamespace(controlsAllowed=True)]) + self.alive = self.valid = self.freq_ok = True + + def all_checks(self, services): + self.asserted_services = services + return self.alive and self.valid and self.freq_ok + + reader = Reader() + with patch.object(messaging, 'SubMaster', return_value=reader) as factory, patch.object(Params, 'get', return_value=True): + inputs = ManualTurnInputs(Params()) + self.assertNotIn('pandaStates', factory.call_args.args[0]) + self.assertEqual(inputs.assist_permission(), ((), False)) + inputs = ManualTurnInputs(Params(), track_assist_permission=True) + self.assertIn('pandaStates', factory.call_args.args[0]) + self.assertEqual(inputs.assist_permission(), (reader['pandaStates'], True)) + for health in ('alive', 'valid', 'freq_ok'): + setattr(reader, health, False) + self.assertEqual(inputs.assist_permission(), (reader['pandaStates'], False)) + setattr(reader, health, True) + self.assertEqual(reader.asserted_services, ['pandaStates']) + + def test_real_card_finalizes_then_binds_only_admitted_controller(self): + harness = preferences_tests.TestVehicleStartupPreferences(methodName='test_only_exact_saved_opt_in_is_loaded') + harness.setUp() + self.addCleanup(harness.doCleanups) + with patch('openpilot.starpilot.controller_extensions.messaging.SubMaster', return_value=SimpleNamespace()): + host, _, published = harness.start(CAR.FORD_MUSTANG_MACH_E_MK1) + self.assertFalse(published.passive) + self.assertIsInstance(host.CI.CC.manual_turn_inputs, ManualTurnInputs) + self.assertIs(host.CI.CC.manual_turn_inputs.params, harness.params) + self.assertTrue(host.CI.CC.manual_turn_inputs.track_assist_permission) + host, _, published = harness.start(CAR.FORD_MUSTANG_MACH_E_MK1, enabled=False) + self.assertTrue(published.passive) + self.assertIsNone(host.CI.CC.manual_turn_inputs) + + def test_standalone_controller_has_no_host_owner(self): + from opendbc.car.ford.tests.test_three_ports import params + from opendbc.car.ford.carcontroller import CarController + from opendbc.car.ford.values import DBC + from opendbc.car.ford.carstate import CarState + from opendbc.car import structs + cp = params(CAR.FORD_MUSTANG_MACH_E_MK1) + controller = CarController(DBC[cp.carFingerprint], cp) + state = CarState(cp) + state.update(state.get_can_parsers(cp)) + state.out = structs.CarState().as_reader() + controller.update(structs.CarControl().as_reader(), state, 0) + self.assertIsNone(controller.manual_turn_inputs) + cp.passive = True + configure_controller(SimpleNamespace(CP=cp, CC=controller), SimpleNamespace()) + self.assertIsNone(controller.manual_turn_inputs) diff --git a/openpilot/starpilot/tests/test_ev6_startup.py b/openpilot/starpilot/tests/test_ev6_startup.py new file mode 100644 index 0000000000..d5b7095e13 --- /dev/null +++ b/openpilot/starpilot/tests/test_ev6_startup.py @@ -0,0 +1,194 @@ +"""Timestamped synthetic CAN/UDS through the actual prepublication caller.""" +import time +import unittest +from unittest.mock import patch + +from opendbc.can.packer import CANPacker +from opendbc.car import Bus, CanData, gen_empty_fingerprint, structs +from opendbc.car import car_helpers, disable_ecu as disable_module +from opendbc.car.hyundai import ecu_startup +from opendbc.car.hyundai.ecu_startup import Outcome, stock_copy +from opendbc.car.hyundai.ev6_startup import EV6Startup, eligible +from opendbc.car.hyundai.hyundaicanfd import hkg_can_fd_checksum +from opendbc.car.hyundai.interface import CarInterface +from opendbc.car.hyundai.ioniq6_handoff import TimestampedCanPacket +from opendbc.car.hyundai.values import CAR, DBC +from openpilot.starpilot.vehicle_startup import VehicleStartupOwner + + +def params(alpha=True, radar=False, car=CAR.KIA_EV6): + fp = gen_empty_fingerprint() + fp[2][0x50] = 16 + fp[1][0x1cf] = 8 + if radar: + if car == CAR.GENESIS_GV70_ELECTRIFIED_1ST_GEN: + fp[0].update(dict.fromkeys(range(0x210, 0x220), 32)) + else: + fp[1][0x500] = 8 + fw = [structs.CarParams.CarFw(ecu=structs.CarParams.Ecu.adas)] + return CarInterface.get_params(car, fp, fw, alpha, False, False) + + +class TestEV6Startup(unittest.TestCase): + @patch.object(time, 'CLOCK_BOOTTIME', getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC), create=True) + def exercise(self, mode='sent', *, capture=True, admission=lambda: True, requested=True, wrong_bus=False, + missing_source=None, fail_ctor=False, warm_fault=None, car=CAR.KIA_EV6): + original = params(car=car) + holder = VehicleStartupOwner() + ci_ref = [] + requests = [] + self.last_requests = requests + packer = CANPacker(DBC[car][Bus.pt]) + counter = 0 + def recv(wait_for_one=False): + nonlocal counter + counter += 1 + if not ci_ref: + if not capture: + return [] + data = bytearray(32) + data[2], data[4] = counter % 256, 7 + data[:2] = hkg_can_fd_checksum(0x51, None, data).to_bytes(2, 'little') + frames = [CanData(0x51, bytes(data), 1 if wrong_bus else 0)] + else: + frames = [] + for parser in ci_ref[0].can_parsers.values(): + for address in parser.addresses: + name = parser.dbc.addr_to_msg[address].name + if name != missing_source and name != 'CRUISE_BUTTONS_ALT': + frame = CanData(*packer.make_can_msg(name, 0 if warm_fault == 'wrongbus' and name == 'SCC_CONTROL' else parser.bus, {})) + frames.extend([frame] * (7 if warm_fault == 'counter' and name == 'SCC_CONTROL' else 1)) + stamp = 1 if ci_ref and warm_fault == 'stale' else time.clock_gettime_ns(time.CLOCK_BOOTTIME) + return [TimestampedCanPacket(frames, stamp)] + def query(send, can_recv, bus, addresses, commands, responses, **kwargs): + command = commands[0] + self.assertEqual(bus, 1) + self.assertEqual(addresses, [(0x730, None)]) + self.assertEqual(responses, [{b'\x10\x03': b'\x50\x03', b'\x28\x83\x01': b'', b'\x28\x00\x01': b'\x68\x00'}[command]]) + requests.append(command) + class Query: + def get_data(self, timeout): + send([CanData(0x730, bytes((len(command),))+command+bytes(7-len(command)), 1)]) + if command == b'\x28\x83\x01' and mode not in ('sent', 'sent_restore'): + raise OSError('simulated partial send') + if command == b'\x28\x00\x01' and mode not in ('restored', 'sent_restore'): + return {} + return {(0x730, None): b''} + return Query() + def hook(cp, candidate, fingerprints, firmware): + return holder.prepare(cp, CarInterface, (recv, lambda frames: None), requested=requested, admission=admission) + fp = gen_empty_fingerprint() + fp[2][0x50], fp[1][0x1cf] = 16, 8 + fw = [structs.CarParams.CarFw(ecu=structs.CarParams.Ecu.adas)] + fingerprint = (car, fp, '0'*17, fw, original.fingerprintSource, True) + with patch.object(car_helpers, 'fingerprint', return_value=fingerprint), \ + patch.object(disable_module, 'IsoTpParallelQuery', side_effect=query), \ + patch.object(ecu_startup, 'IsoTpParallelQuery', side_effect=query): + if fail_ctor: + try: + with patch.object(CarInterface, 'CarController', side_effect=RuntimeError('constructor failed')): + car_helpers.get_car(recv, lambda frames: None, lambda *args: None, True, False, pre_create_hook=hook) + finally: + holder.close() + ci = car_helpers.get_car(recv, lambda frames: None, lambda *args: None, True, False, pre_create_hook=hook) + ci_ref.append(ci) + holder.configure(ci) + holder.seal_publication() + return ci, holder, requests + + def test_sent_prepublication_binds_immutable_template_and_suppressed_protocol(self): + ci, holder, requests = self.exercise() + self.assertIs(holder.owner.outcome, Outcome.SENT_UNCONFIRMED) + self.assertIs(ci.CC.adrv_template, holder.owner.template) + self.assertTrue(ci.CP.openpilotLongitudinalControl) + self.assertIn(b'\x28\x83\x01', requests) + self.assertNotIn(b'\x28\x03\x01', requests) + ci.CP.safetyConfigs[-1].safetyParam ^= 4 + with self.assertRaisesRegex(RuntimeError, 'prepared'): + holder.seal_publication() + + def test_missing_capture_never_requests_uds_and_partial_restore_warms_actual_stock(self): + ci, holder, requests = self.exercise(capture=False) + self.assertEqual(requests, []) + self.assertIs(holder.owner.outcome, Outcome.STOCK_UNTOUCHED) + self.assertFalse(ci.CP.openpilotLongitudinalControl) + ci, holder, requests = self.exercise('restored') + self.assertIs(holder.owner.outcome, Outcome.STOCK_RESTORED) + self.assertIn(b'\x28\x00\x01', requests) + self.assertTrue(holder.owner.ready) + self.assertIsNone(ci.CC.adrv_template) + + def test_partial_failure_without_restore_admission_cannot_publish(self): + admitted = iter((True, False)) + with self.assertRaisesRegex(RuntimeError, 'restoration unverified'): + self.exercise('restored', admission=lambda: next(admitted)) + with self.assertRaisesRegex(RuntimeError, 'restoration unverified'): + self.exercise('uncertain') + + def test_stock_copy_exact_and_button_pause_policy_and_explicit_owner_guard(self): + for radar in (False, True): + self.assertEqual(stock_copy(params(radar=radar)).to_dict(), params(False, radar).to_dict()) + ci = CarInterface(params()) + buttons = ci.can_parsers[Bus.pt].message_states[0x1cf] + self.assertGreaterEqual(buttons.timeout_threshold, 500_000_000) + self.assertNotIn(0x1a0, ci.can_parsers[Bus.pt].addresses) + stock = CarInterface(params(False)) + self.assertIn(0x1a0, stock.can_parsers[Bus.pt].addresses) + with self.assertRaisesRegex(RuntimeError, 'matching prepared'): + VehicleStartupOwner().configure(ci) + self.assertIsNone(CarInterface.startup_owner(params(), (list, lambda frames: None), requested=False)) + sibling = params() + sibling.carFingerprint = CAR.HYUNDAI_IONIQ_6 + self.assertFalse(eligible(sibling)) + + def test_missing_stock_source_cannot_publish_and_buttons_can_pause(self): + with self.assertRaisesRegex(RuntimeError, 'stock CANFD'): + self.exercise('restored', missing_source='SCC_CONTROL') + ci, holder, _ = self.exercise('restored') + parser = ci.can_parsers[Bus.pt] + # Freeze the real last button receipt while refreshing all other sources. + packer = CANPacker(DBC[CAR.KIA_EV6][Bus.pt]) + stamp = max(source.timestamps[-1] for source in parser.message_states.values() if source.timestamps) + for i in range(1, 51): + frames = [] + for source_parser in ci.can_parsers.values(): + for address in source_parser.addresses: + name = source_parser.dbc.addr_to_msg[address].name + if name not in ('CRUISE_BUTTONS', 'CRUISE_BUTTONS_ALT'): + frames.append(CanData(*packer.make_can_msg(name, source_parser.bus, {}))) + ci.update([(stamp+i*10_000_000, frames)]) + self.assertTrue(ci.CS.out.canValid) + self.assertTrue(holder.owner._sources(ci, stamp+500_000_000, floor=holder.owner.floor_ns)) + + def test_capture_rejects_wrong_bus_crc_repeats_and_old_producers(self): + for bad in ('bus', 'crc', 'repeat', 'stale'): + with self.subTest(bad=bad): + tick = [1_000_000_000] + counter = [0] + def recv(tick=tick, counter=counter, bad=bad): + tick[0] += 100_000_000 + counter[0] += 1 + data = bytearray(32) + data[2] = 7 if bad == 'repeat' else counter[0] + data[4] = 1 + data[:2] = hkg_can_fd_checksum(0x51, None, data).to_bytes(2, 'little') + if bad == 'crc': + data[0] ^= 1 + return [TimestampedCanPacket([CanData(0x51, bytes(data), 1 if bad == 'bus' else 0)], + 1 if bad == 'stale' else tick[0])] + owner = EV6Startup(params(), (recv, lambda frames: self.fail('capture must not transmit'))) + with patch.object(time, 'CLOCK_BOOTTIME', getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC), create=True), \ + patch.object(time, 'clock_gettime_ns', side_effect=lambda clock, tick=tick: tick[0]), \ + patch.object(time, 'monotonic', side_effect=lambda tick=tick: tick[0]/1e9), patch.object(time, 'sleep'): + self.assertIsNone(owner._capture()) + + def test_unpublished_actual_constructor_failure_restores_transaction(self): + with self.assertRaisesRegex(RuntimeError, 'constructor failed'): + self.exercise('sent_restore', fail_ctor=True) + self.assertIn(b'\x28\x83\x01', self.last_requests) + self.assertIn(b'\x28\x00\x01', self.last_requests) + + def test_wrong_bus_stale_and_counter_loss_stock_sources_cannot_publish(self): + for fault in ('wrongbus', 'stale', 'counter'): + with self.subTest(fault=fault), self.assertRaisesRegex(RuntimeError, 'stock CANFD'): + self.exercise('restored', warm_fault=fault) diff --git a/openpilot/starpilot/tests/test_g90_lead_inputs.py b/openpilot/starpilot/tests/test_g90_lead_inputs.py new file mode 100644 index 0000000000..27e23783a5 --- /dev/null +++ b/openpilot/starpilot/tests/test_g90_lead_inputs.py @@ -0,0 +1,169 @@ +"""Actual scoped controller/provider callers with explicit synthetic radar transport.""" +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +from opendbc.can import CANParser +from opendbc.car import Bus, structs +from opendbc.car.hyundai.g90_lead import G90LeadState +from opendbc.car.hyundai.interface import CarInterface +from opendbc.car.hyundai.tests.test_g90_lead import observation +from opendbc.car.hyundai.tests.test_g90_longitudinal import params +from opendbc.car.hyundai.values import CAR, DBC +from openpilot.starpilot.controller_extensions import configure_controller +from openpilot.starpilot.longitudinal import g90_lead, radar_lead_context + + +class RadarMessages(dict): + def __init__(self, auto=False): + super().__init__(radarState=SimpleNamespace( + leadOne=SimpleNamespace(present=True, dRel=26., vRel=-.5), + radarErrors=SimpleNamespace(canError=False, radarFault=False, wrongConfig=False, radarUnavailableTemporary=False)), + carState=SimpleNamespace(canValid=True, canTimeout=False), + deviceState=SimpleNamespace(started=True, startedMonoTime=500_000_000)) + self.now = 1_000_000_000 + self.auto = auto + self.seen = dict.fromkeys(self, True) + self.alive = dict.fromkeys(self, True) + self.valid = dict.fromkeys(self, True) + self.logMonoTime = {} + self.recv_time = {} + self.refresh(*self) + + def refresh(self, *names): + for name in names: + self.logMonoTime[name] = self.now - 1000 + self.recv_time[name] = (self.now - 500) / 1e9 + + def update(self, *args, **kwargs): + if self.auto: + self.now += 10_000_000 + self.refresh(*self) + + +class TestG90LeadInputs(unittest.TestCase): + def provider(self, messages): + with patch.object(radar_lead_context.messaging, 'SubMaster', return_value=messages) as constructor: + provider = g90_lead.G90LeadInputs() + constructor.assert_called_once_with(['radarState', 'deviceState', 'carState']) + return provider + + def sample(self, provider, messages): + with patch.object(radar_lead_context, 'clock_pair_ns', side_effect=lambda: (messages.now, messages.now+20_000_000_000)): + return provider.update() + + def test_staggered_producers_recover_without_a_moving_floor(self): + messages = RadarMessages() + provider = self.provider(messages) + self.assertIsNone(self.sample(provider, messages)) + floor = provider.floor_ns + for service in ('deviceState', 'carState', 'radarState'): + messages.now += 10_000_000 + messages.refresh(service) + result = self.sample(provider, messages) + self.assertEqual(provider.floor_ns, floor) + assert result is not None + messages.now += 10_000_000 + assert self.sample(provider, messages) is not None + messages.valid['radarState'] = False + self.assertIsNone(self.sample(provider, messages)) + revoked_floor = provider.floor_ns + messages.valid['radarState'] = True + for service in ('radarState', 'carState', 'deviceState'): + messages.now += 10_000_000 + messages.refresh(service) + result = self.sample(provider, messages) + self.assertEqual(provider.floor_ns, revoked_floor) + assert result is not None + + def test_stale_invalid_and_expected_transport_errors_are_optional(self): + for failure in ('stale', 'future', 'receipt', 'radar_fault', 'nonfinite', 'transport'): + messages = RadarMessages(auto=True) + provider = self.provider(messages) + self.sample(provider, messages) + assert self.sample(provider, messages) is not None + messages.auto = False + if failure == 'stale': + messages.now += 150_000_001 + elif failure == 'future': + messages.logMonoTime['radarState'] = messages.now+1 + elif failure == 'receipt': + messages.recv_time['radarState'] = (messages.now+1)/1e9 + elif failure == 'radar_fault': + messages['radarState'].radarErrors.radarFault = True + elif failure == 'nonfinite': + messages['radarState'].leadOne.vRel = float('nan') + else: + vars(messages)['update'] = lambda timeout: (_ for _ in ()).throw(OSError('explicit transport failure')) + self.assertIsNone(self.sample(provider, messages), failure) + ci = CarInterface(params()) + with patch.object(radar_lead_context.messaging, 'SubMaster', side_effect=OSError('explicit constructor failure')), \ + patch('openpilot.starpilot.controller_extensions.cloudlog.exception') as diagnostic: + configure_controller(ci, None) + self.assertIsNone(ci.CC.g90_lead_inputs) # Existing fallback wire, not fresh lead evidence. + diagnostic.assert_called_once() + state = G90LeadState() + for _ in range(50): + state.update(observation(), False) + self.assertFalse(state.update(None, True).lead_visible) # Missing source is not a fake20m lead. + self.assertFalse(state.update(observation(drive=2), False).lead_visible) + + def test_provider_clock_and_drive_epochs_require_new_staggered_sources(self): + messages = RadarMessages(auto=True) + provider = self.provider(messages) + self.sample(provider, messages) + assert self.sample(provider, messages) is not None + messages.auto = False + offset = 40_000_000_000 + def sample_epoch(): + with patch.object(radar_lead_context, 'clock_pair_ns', side_effect=lambda: (messages.now, messages.now+offset)): + return provider.update() + self.assertIsNone(sample_epoch()) # Suspend changed BOOTTIME/MONOTONIC offset. + floor = provider.floor_ns + self.assertIsNone(sample_epoch()) # The old cache cannot reauthorize. + for service in ('deviceState', 'radarState', 'carState'): + messages.now += 10_000_000 + messages.refresh(service) + result = sample_epoch() + self.assertEqual(provider.floor_ns, floor) + assert result is not None + messages.now += 10_000_000 + messages['deviceState'].startedMonoTime = messages.now-2000 + messages.refresh('deviceState') + self.assertIsNone(sample_epoch()) # A new drive cannot reuse old radar/car. + floor = provider.floor_ns + for service in ('radarState', 'carState', 'deviceState'): + messages.now += 10_000_000 + messages.refresh(service) + result = sample_epoch() + self.assertEqual(provider.floor_ns, floor) + assert result is not None + + def test_actual_controller_provider_bind_and_100hz_to_50hz_wire(self): + cp = params() + ci = CarInterface(cp) + ci.update([]) + messages = RadarMessages(auto=True) + with patch.object(radar_lead_context.messaging, 'SubMaster', return_value=messages): + configure_controller(ci, None) + self.assertIsInstance(ci.CC.g90_lead_inputs, g90_lead.G90LeadInputs) + command = structs.CarControl(enabled=True, longActive=True) + command.actuators.longControlState = structs.CarControl.Actuators.LongControlState.pid + parser = CANParser(DBC[CAR.GENESIS_G90][Bus.pt], [('SCC11', 0), ('SCC14', 0)], 0) + obj = parser.dbc.name_to_msg['SCC14'].sigs['ObjDistStat'] + self.assertEqual((obj.start_bit, obj.size, obj.factor, obj.offset), (42, 2, 1, 0)) + captures = {} + for frame in range(52): + with patch.object(radar_lead_context, 'clock_pair_ns', side_effect=lambda: (messages.now, messages.now+20_000_000_000)): + _, packets = ci.apply(command.as_reader(), (frame+1)*10_000_000) + scc = [packet for packet in packets if packet[0] in (0x420, 0x389)] + self.assertEqual(bool(scc), frame % 2 == 0) + if scc: + parser.update([(1_000_000_000+frame*10_000_000, scc)]) + captures[frame] = (parser.vl['SCC11']['ObjValid'], parser.vl['SCC14']['ObjGap']) + self.assertEqual(captures[48], (0, 0)) + self.assertEqual(captures[50], (1, 4)) + self.assertEqual(parser.vl['SCC11']['ACC_ObjDist'], 26) + self.assertEqual(parser.vl['SCC11']['ACC_ObjRelSpd'], -.5) + self.assertEqual(parser.vl['SCC14']['ObjDistStat'], 2) + diff --git a/openpilot/starpilot/tests/test_gv70_fallback_cancel.py b/openpilot/starpilot/tests/test_gv70_fallback_cancel.py new file mode 100644 index 0000000000..2087e636a0 --- /dev/null +++ b/openpilot/starpilot/tests/test_gv70_fallback_cancel.py @@ -0,0 +1,30 @@ +"""Requested GV70 takeover fallback keeps stock Cancel and Resume distinct.""" +import unittest + +from opendbc.car import structs +from opendbc.car.hyundai.values import CAR +from openpilot.starpilot.tests import test_ev6_startup as fixture + + +class TestGV70FallbackCancel(unittest.TestCase): + def test_actual_startup_fallback_suppresses_cancel_but_preserves_resume(self): + for mode, capture in (("restored", True), ("sent", False)): + ci, holder, _ = fixture.TestEV6Startup("runTest").exercise( + mode, capture=capture, car=CAR.GENESIS_GV70_ELECTRIFIED_1ST_GEN) + self.assertFalse(ci.CP.openpilotLongitudinalControl) + self.assertTrue(ci.CC.gv70_stock_fallback) + command = structs.CarControl() + command.cruiseControl.cancel = True + cancels = [] + for tick in range(60): + _, frames = ci.apply(command.as_reader(), (tick+1)*10_000_000) + cancels.extend(frame for frame in frames if frame[0] == 0x1cf) + self.assertEqual(cancels, []) + command.cruiseControl.cancel = False + command.cruiseControl.resume = True + resumes = [] + for tick in range(60, 120): + _, frames = ci.apply(command.as_reader(), (tick+1)*10_000_000) + resumes.extend(frame for frame in frames if frame[0] == 0x1cf) + self.assertTrue(resumes) + self.assertTrue(holder.owner.ready) diff --git a/openpilot/starpilot/tests/test_gv70_lead_inputs.py b/openpilot/starpilot/tests/test_gv70_lead_inputs.py new file mode 100644 index 0000000000..8c4240c884 --- /dev/null +++ b/openpilot/starpilot/tests/test_gv70_lead_inputs.py @@ -0,0 +1,192 @@ +"""Joined optional lead callers with explicit synthetic CAN and transport clocks.""" +import time +import unittest +from unittest.mock import patch + +from opendbc.can.packer import CANPacker +from opendbc.can.parser import CANParser +from opendbc.car import Bus, CanData, structs +from opendbc.car.hyundai.canfd_lead import ABSENT +from opendbc.car.hyundai.gv70_camera_lead import GV70CameraLead +from opendbc.car.hyundai.hyundaicanfd import CanBus, create_acc_control +from opendbc.car.hyundai.interface import CarInterface +from opendbc.car.hyundai.tests.test_gv70_camera_lead import packet as camera_packet, params +from opendbc.car.hyundai.values import CAR, DBC +from openpilot.starpilot.controller_extensions import configure_controller +from openpilot.starpilot.longitudinal import radar_lead_context +from openpilot.starpilot.longitudinal.gv70_lead import GV70LeadInputs +from openpilot.starpilot.tests.test_g90_lead_inputs import RadarMessages + +OFFSET = 20_000_000_000 + + +class TestGV70LeadInputs(unittest.TestCase): + def setUp(self): + # Mac has no BOOTTIME constant. Both synthetic CAN and paired-clock values + # explicitly use the same chosen BOOTTIME domain in these host fixtures. + self.boot = patch.object(time, 'CLOCK_BOOTTIME', getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC), create=True) + self.boot.start() + self.addCleanup(self.boot.stop) + + def provider(self, messages): + with patch.object(radar_lead_context.messaging, 'SubMaster', return_value=messages): + return GV70LeadInputs() + + def sample(self, provider, messages, camera=None, hud=False, control=None, offset=OFFSET): + with patch.object(radar_lead_context, 'clock_pair_ns', side_effect=lambda: (messages.now, messages.now+offset)): + return provider.update(camera, hud, messages.now if control is None else control) + + def camera(self, messages, *, offset=OFFSET, distance=12., relative=-2.): + cp = params() + owner = GV70CameraLead(cp) + packer = CANPacker(DBC[cp.carFingerprint][Bus.pt]) + with patch.object(time, 'clock_gettime_ns', return_value=messages.now+offset): + owner.update([(messages.now+offset-1000, [camera_packet(packer, owner.bus, 1, distance, relative)])]) + return owner + + def test_camera_recovery_staggered_context_is_independent_of_missing_radar(self): + messages = RadarMessages() + messages.seen['radarState'] = False + provider = self.provider(messages) + self.assertEqual(self.sample(provider, messages), ABSENT) + floor = provider.context.floor_ns + for name in ('deviceState', 'carState'): + messages.now += 10_000_000 + messages.refresh(name) + camera = self.camera(messages) + result = self.sample(provider, messages, camera) + self.assertEqual(provider.context.floor_ns, floor) + self.assertTrue(result.visible) + self.assertEqual(result.distance, 12.) + messages.now += 10_000_000 + self.assertTrue(self.sample(provider, messages, camera).visible) + messages.seen['radarState'] = True + messages['radarState'].radarErrors.radarFault = True + self.assertEqual(self.sample(provider, messages, camera).distance, 12.) + self.assertEqual(provider.context.floor_ns, floor) + + def test_radar_precedence_controls_and_clock_drive_epochs_reject_old_sources(self): + messages = RadarMessages(auto=True) + provider = self.provider(messages) + self.sample(provider, messages) + self.assertEqual(self.sample(provider, messages).distance, 26.) + camera = self.camera(messages) + messages.valid['radarState'] = False + self.assertEqual(self.sample(provider, messages, camera).distance, 12.) + self.assertEqual(self.sample(provider, messages, None, True).distance, 20.) + self.assertEqual(self.sample(provider, messages, camera, control=messages.now+1_000_000_000), ABSENT) + self.assertEqual(self.sample(provider, messages, camera, control=messages.now-150_000_001), ABSENT) + messages.auto = False + messages.now += 10_000_000 + messages.refresh('deviceState', 'carState') + self.assertEqual(self.sample(provider, messages, camera, offset=OFFSET+20_000_000_000), ABSENT) + floor = provider.context.floor_ns + for name in ('deviceState', 'carState'): + messages.now += 10_000_000 + messages.refresh(name) + self.assertEqual(self.sample(provider, messages, camera, offset=OFFSET+20_000_000_000), ABSENT) + self.assertEqual(provider.context.floor_ns, floor) + fresh_camera = self.camera(messages, offset=OFFSET+20_000_000_000) + self.assertTrue(self.sample(provider, messages, fresh_camera, offset=OFFSET+20_000_000_000).visible) + messages['deviceState'].startedMonoTime = messages.now-1 + messages.now += 10_000_000 + messages.refresh('deviceState', 'carState') + self.assertEqual(self.sample(provider, messages, fresh_camera, offset=OFFSET+20_000_000_000), ABSENT) + for name in ('carState', 'deviceState'): + messages.now += 10_000_000 + messages.refresh(name) + self.assertEqual(self.sample(provider, messages, fresh_camera, offset=OFFSET+20_000_000_000), ABSENT) + new_camera = self.camera(messages, offset=OFFSET+20_000_000_000) + self.assertTrue(self.sample(provider, messages, new_camera, offset=OFFSET+20_000_000_000).visible) + + def test_actual_interface_isolated_camera_bad_counter_keeps_required_health(self): + for lka in (True, False): + cp = params(lka) + ci, baseline = CarInterface(cp), CarInterface(params(lka)) + ci.update([]) + baseline.update([]) + packer = CANPacker(DBC[cp.carFingerprint][Bus.pt]) + camera = ci.CS.gv70_camera_lead + good = camera_packet(packer, camera.bus, 1) + for tick in range(60): + stamp = 1_000_000_000+tick*10_000_000 + frames = [] + for parser in ci.can_parsers.values(): + self.assertNotIn(0x1b5, parser.addresses) + for address in parser.addresses: + name = parser.dbc.addr_to_msg[address].name + frames.append(CanData(*packer.make_can_msg(name, parser.bus, {}))) + with patch.object(time, 'clock_gettime_ns', return_value=stamp): + actual = ci.update([(stamp, frames+[good])]) + expected = baseline.update([(stamp, frames)]) + self.assertEqual(actual.canValid, expected.canValid) + self.assertEqual(actual.canTimeout, expected.canTimeout) + self.assertTrue(actual.canValid) + self.assertEqual(camera.observation.producer_boot_ns, 1_000_000_000) + self.assertIsNone(camera.current(stamp)) + self.assertNotIn(camera.parser, ci.can_parsers.values()) + + def test_actual_configure_camera_context_and_50hz_packed_lead_have_no_hysteresis(self): + for lka in (True, False): + cp = params(lka) + ci = CarInterface(cp) + messages = RadarMessages() + messages.seen['radarState'] = False + with patch.object(radar_lead_context.messaging, 'SubMaster', return_value=messages): + configure_controller(ci, None) + self.assertIsInstance(ci.CC.gv70_lead_inputs, GV70LeadInputs) + packer = CANPacker(DBC[cp.carFingerprint][Bus.pt]) + decoder = CANParser(DBC[cp.carFingerprint][Bus.pt], [('SCC_CONTROL', 50)], CanBus(cp).ECAN) + command = structs.CarControl() + command.enabled = True + command.actuators.accel = 1. + ci.update([]) + values_by_tick = {} + for tick in range(8): + messages.now += 10_000_000 + messages.refresh('deviceState', 'carState') + boot_stamp = messages.now+OFFSET-1000 + frames = [] + for parser in ci.can_parsers.values(): + for address in parser.addresses: + name = parser.dbc.addr_to_msg[address].name + frames.append(CanData(*packer.make_can_msg(name, parser.bus, {}))) + frames.append(camera_packet(packer, ci.CS.gv70_camera_lead.bus, tick+1, 30., -1.)) + with patch.object(time, 'clock_gettime_ns', return_value=messages.now+OFFSET): + state = ci.update([(boot_stamp, frames)]) + control_now = messages.now + with patch.object(radar_lead_context, 'clock_pair_ns', side_effect=lambda messages=messages: (messages.now, messages.now+OFFSET)): + _, outputs = ci.apply(command.as_reader(), control_now) + packets = [CanData(*frame) for frame in outputs if frame[0] == 0x1a0] + self.assertEqual(len(packets), int(tick % 2 == 0)) + if packets: + decoder.update([(boot_stamp, packets)]) + values_by_tick[tick] = dict(decoder.vl['SCC_CONTROL']) + self.assertTrue(state.canValid) + self.assertFalse(state.canTimeout) + self.assertEqual(values_by_tick[0]['ACC_ObjDist'], 0.) + self.assertEqual(values_by_tick[0]['ObjValid'], 1) + self.assertEqual(values_by_tick[2]['ACC_ObjDist'], 30.) + self.assertAlmostEqual(values_by_tick[2]['ACC_ObjRelSpd'], -1., places=12) + self.assertEqual(values_by_tick[2]['ObjValid'], 0) + self.assertEqual(values_by_tick[2]['OBJ_STATUS'], 2) + self.assertEqual(values_by_tick[2]['JerkUpperLimit'], 1.5) + self.assertAlmostEqual(values_by_tick[2]['aReqValue'], .06) + + def test_optional_constructor_failure_and_unchanged_generic_helper_bytes(self): + ci = CarInterface(params()) + with patch.object(radar_lead_context.messaging, 'SubMaster', side_effect=OSError('explicit constructor failure')), \ + patch('openpilot.starpilot.controller_extensions.cloudlog.exception') as diagnostic: + configure_controller(ci, None) + self.assertIsNone(ci.CC.gv70_lead_inputs) + diagnostic.assert_called_once() + for car in (CAR.KIA_EV6, CAR.HYUNDAI_IONIQ_6): + cp = params() + cp.carFingerprint = car + old, new = CANPacker(DBC[car][Bus.pt]), CANPacker(DBC[car][Bus.pt]) + arguments = (CanBus(cp), True, 0., .5, False, False, 80., structs.CarControl.HUDControl()) + expected_values = {'ACCMode': 1, 'MainMode_ACC': 1, 'StopReq': 0, 'aReqValue': .1, 'aReqRaw': .5, + 'VSetDis': 80., 'JerkLowerLimit': 5., 'JerkUpperLimit': 3., 'ACC_ObjDist': 1, + 'ObjValid': 0, 'OBJ_STATUS': 2, 'SET_ME_2': 4, 'SET_ME_3': 3, 'SET_ME_TMP_64': 100, + 'DISTANCE_SETTING': 0} + self.assertEqual(old.make_can_msg('SCC_CONTROL', CanBus(cp).ECAN, expected_values), create_acc_control(new, *arguments)) diff --git a/openpilot/starpilot/tests/test_gv70_startup.py b/openpilot/starpilot/tests/test_gv70_startup.py new file mode 100644 index 0000000000..6331f3b828 --- /dev/null +++ b/openpilot/starpilot/tests/test_gv70_startup.py @@ -0,0 +1,89 @@ +"""Exact first-generation GV70 startup with synthetic captured-CAN evidence.""" +import unittest + +from opendbc.can.packer import CANPacker +from opendbc.car import Bus, CanData, structs +from opendbc.car.hyundai.ecu_startup import Outcome, stock_copy +from opendbc.car.hyundai.ev6_template import EV6Template +from opendbc.car.hyundai.gv70_startup import eligible +from opendbc.car.hyundai.gv70_template import GV70Template +from opendbc.car.hyundai.interface import CarInterface +from opendbc.car.hyundai.values import CAR, DBC +from openpilot.starpilot.tests import test_ev6_startup as fixture +from openpilot.starpilot.vehicle_startup import VehicleStartupOwner +from opendbc.car.hyundai.tests.test_gv70_template import capture + + +CAR_ID = CAR.GENESIS_GV70_ELECTRIFIED_1ST_GEN + + +class TestGV70Startup(unittest.TestCase): + def exercise(self, *args, **kwargs): + harness = fixture.TestEV6Startup('runTest') + return harness.exercise(*args, car=CAR_ID, **kwargs) + + def test_actual_prepublication_capture_type_cp_and_suppressed_protocol(self): + ci, holder, requests = self.exercise() + self.assertIs(holder.owner.outcome, Outcome.SENT_UNCONFIRMED) + self.assertIsInstance(ci.CC.adrv_template, GV70Template) + self.assertNotIsInstance(ci.CC.adrv_template, EV6Template) + self.assertEqual(ci.CP.safetyConfigs[-1].safetyParam, 0x15) + self.assertIn(b'\x28\x83\x01', requests) + self.assertNotIn(b'\x28\x03\x01', requests) + ci.CP.safetyConfigs[-1].safetyParam ^= 4 + with self.assertRaisesRegex(RuntimeError, 'prepared'): + holder.seal_publication() + + def test_stock_fallback_and_full_radar_present_absent_cp_equality(self): + for radar in (False, True): + candidate, stock = fixture.params(radar=radar, car=CAR_ID), fixture.params(False, radar, CAR_ID) + self.assertEqual(stock_copy(candidate).to_dict(), stock.to_dict()) + self.assertEqual(candidate.radarUnavailable, not radar) + for mode, has_capture, expected in (('sent', False, Outcome.STOCK_UNTOUCHED), + ('restored', True, Outcome.STOCK_RESTORED)): + ci, holder, requests = self.exercise(mode, capture=has_capture) + self.assertIs(holder.owner.outcome, expected) + self.assertFalse(ci.CP.openpilotLongitudinalControl) + self.assertEqual(ci.CP.safetyConfigs[-1].safetyParam, 0x11) + self.assertIsNone(ci.CC.adrv_template) + self.assertEqual(b'\x28\x00\x01' in requests, has_capture) + if not has_capture: + self.assertEqual(requests, []) + + def test_uncertain_failure_required_owner_and_topology_exclusions(self): + with self.assertRaisesRegex(RuntimeError, 'restoration unverified'): + self.exercise('uncertain') + candidate = fixture.params(car=CAR_ID) + self.assertTrue(eligible(candidate)) + with self.assertRaisesRegex(RuntimeError, 'matching prepared'): + VehicleStartupOwner().configure(CarInterface(candidate)) + self.assertIsNone(CarInterface.startup_owner(candidate, (list, lambda frames: None), requested=False)) + candidate.carFingerprint = CAR.GENESIS_GV70_ELECTRIFIED_2ND_GEN + self.assertFalse(eligible(candidate)) + candidate.carFingerprint = CAR.GENESIS_GV70_1ST_GEN + self.assertFalse(eligible(candidate)) + + def test_actual_wheel_source_to_ordered_controller_gv70_speed_and_ev6_preservation(self): + for car, template_type in ((CAR_ID, GV70Template), (CAR.KIA_EV6, EV6Template)): + cp = fixture.params(car=car) + ci = CarInterface(cp) + ci.CC.adrv_template = template_type.capture(capture(7)) + packer = CANPacker(DBC[car][Bus.pt]) + command = structs.CarControl() + command.enabled = True + outputs = [] + for tick in range(4): + # Actual four-wheel source feeds CarState raw speed before its KF. + wheels = dict.fromkeys(('WHL_SpdFLVal', 'WHL_SpdFRVal', 'WHL_SpdRLVal', 'WHL_SpdRRVal'), 36.) + ci.update([((tick+1)*10_000_000, [CanData(*packer.make_can_msg('WHEEL_SPEEDS', 1, wheels)), + CanData(*packer.make_can_msg('ACCELERATOR', 1, {'GEAR': 4})), + CanData(*packer.make_can_msg('CAM_0x2a4', 2, {}))])]) + _, frames = ci.apply(command.as_reader(), (tick+1)*10_000_000) + outputs.extend(CanData(*frame) for frame in frames if frame[0] == 0x51) + self.assertAlmostEqual(ci.CS.out.vEgoRaw, 10.*cp.wheelSpeedFactor) + self.assertEqual(len(outputs), 4) + if car == CAR_ID: + self.assertEqual(int.from_bytes(outputs[-1].dat[8:10], 'little'), round(ci.CS.out.vEgoRaw*100)) + else: + self.assertEqual(outputs[-1].dat[8:10], capture(7)[8:10]) + self.assertTrue(all(frame.src == 0 and len(frame.dat) == 32 for frame in outputs)) diff --git a/openpilot/starpilot/tests/test_vehicle_preferences.py b/openpilot/starpilot/tests/test_vehicle_preferences.py new file mode 100644 index 0000000000..114b7e5b3f --- /dev/null +++ b/openpilot/starpilot/tests/test_vehicle_preferences.py @@ -0,0 +1,124 @@ +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import Mock, patch + +from opendbc.car import car_helpers, gen_empty_fingerprint, structs +from opendbc.car.toyota.values import CAR, ToyotaFlags +from opendbc.safety import ALTERNATIVE_EXPERIENCE +from openpilot.common.params import Params +from openpilot.selfdrive.car import card +from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences + + +class TestVehicleStartupPreferences(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.params = Params(temporary.name) + + def raw(self, key, value): + path = Path(self.params.get_param_path(key)) + if value is None: + path.unlink(missing_ok=True) + else: + path.write_bytes(value) + + def test_suburban_saved_long_pitch_configures_exact_controller(self): + from opendbc.car.gm.carcontroller import CarController + from opendbc.car.gm.interface import CarInterface + from opendbc.car.gm.values import CAR as GM_CAR, DBC + fingerprint = gen_empty_fingerprint() + fingerprint[1][0x460] = 8 + cp = CarInterface.get_params(GM_CAR.CHEVROLET_SUBURBAN, fingerprint, [], False, False, False) + for requested in (False, True): + self.raw("LongPitch", b"1" if requested else b"0") + preferences = VehicleStartupPreferences.read(self.params, enabled=True) + controller = CarController(DBC[cp.carFingerprint], cp) + preferences.configure_controller(SimpleNamespace(CP=cp, CC=controller)) + self.assertEqual(controller.long_pitch, requested) + + def test_only_exact_saved_opt_in_is_loaded(self): + for raw in (None, b"0", b"", b"true", b"1\n", b"1" * 20, b"1"): + with self.subTest(raw=raw): + self.raw("ToyotaAutoHold", raw) + preferences = VehicleStartupPreferences.read(self.params, enabled=True) + self.assertEqual(preferences.toyota_auto_hold, raw == b"1") + self.assertFalse(VehicleStartupPreferences.read(self.params, enabled=False).toyota_auto_hold) + for safe in (b"1", b"invalid", b"", b"0", None): + with self.subTest(safe=safe): + self.raw("SafeMode", safe) + self.assertEqual(VehicleStartupPreferences.read(self.params, enabled=True).toyota_auto_hold, safe in (None, b"0")) + self.raw("SafeMode", None) + path = Path(self.params.get_param_path("ToyotaAutoHold")) + path.unlink() + path.symlink_to("SafeMode") + self.assertFalse(VehicleStartupPreferences.read(self.params, enabled=True).toyota_auto_hold) + + def start(self, candidate, *, enabled=True, requested=True, change_saved=False): + self.raw("ToyotaAutoHold", b"1" if requested else b"0") + self.params.put_bool("OpenpilotEnabledToggle", enabled, block=True) + snapshots = [] + + def fingerprint(*args, **kwargs): + return candidate, gen_empty_fingerprint(), "0" * 17, [], structs.CarParams.FingerprintSource.can, True + + def create(*args, **kwargs): + ci = car_helpers.get_car(*args, **kwargs) + snapshots.append((bool(ci.CP.flags & ToyotaFlags.AUTO_BRAKE_HOLD), int(ci.CP.alternativeExperience))) + if change_saved: + self.raw("ToyotaAutoHold", b"0") + return ci + + with patch.object(card, "Params", return_value=self.params), \ + patch.object(card, "feature_requested", return_value=False), \ + patch.object(card.messaging, "sub_sock", return_value=Mock()), \ + patch.object(card.messaging, "SubMaster", return_value=Mock()), \ + patch.object(card.messaging, "PubMaster", return_value=SimpleNamespace(sock={"sendcan": Mock()})), \ + patch.object(card.messaging, "recv_one_retry", return_value=SimpleNamespace(can=[object()])), \ + patch.object(card, "Ratekeeper", return_value=Mock()), \ + patch.object(card, "get_cache", return_value=None), \ + patch.object(card, "put_cache"), \ + patch.object(card, "get_car", side_effect=create), \ + patch.object(car_helpers, "fingerprint", side_effect=fingerprint): + host = card.Car() + with structs.CarParams.from_bytes(self.params.get("CarParams")) as cp: + published = cp.as_builder() + return host, snapshots, published + + def test_real_card_admits_before_construction_and_publishes_final_permission(self): + for candidate, permission in ((CAR.TOYOTA_COROLLA_TSS2, ALTERNATIVE_EXPERIENCE.TOYOTA_AUTO_HOLD), + (CAR.TOYOTA_CAMRY_TSS2, ALTERNATIVE_EXPERIENCE.TOYOTA_AEB_HOLD)): + with self.subTest(candidate=candidate): + host, constructed, published = self.start(candidate, change_saved=True) + self.assertEqual(constructed, [(True, permission)]) + self.assertTrue(host.CI.CS.CP.flags & ToyotaFlags.AUTO_BRAKE_HOLD) + self.assertEqual(published.alternativeExperience, permission) + self.assertTrue(published.flags & ToyotaFlags.AUTO_BRAKE_HOLD) + self.assertEqual(self.params.get("ToyotaAutoHold"), False) + + def test_default_disabled_and_master_off_never_admit_hold(self): + for enabled, requested in ((True, False), (False, True)): + with self.subTest(enabled=enabled, requested=requested): + _, constructed, published = self.start(CAR.TOYOTA_CAMRY_TSS2, enabled=enabled, requested=requested) + self.assertEqual(constructed, [(False, 0)]) + self.assertEqual(published.alternativeExperience, 0) + self.assertFalse(published.flags & ToyotaFlags.AUTO_BRAKE_HOLD) + if not enabled: + self.assertTrue(published.passive) + self.assertEqual(published.safetyConfigs[0].safetyModel, structs.CarParams.SafetyModel.noOutput) + + def test_finalize_cannot_late_enable_unprepared_parser(self): + cp = car_helpers.interfaces[CAR.TOYOTA_CAMRY_TSS2].get_non_essential_params(CAR.TOYOTA_CAMRY_TSS2) + preferences = VehicleStartupPreferences(True) + preferences.finalize(cp) + self.assertFalse(cp.flags & ToyotaFlags.AUTO_BRAKE_HOLD) + self.assertEqual(cp.alternativeExperience, 0) + + def test_unrelated_vehicle_flags_are_untouched(self): + cp = SimpleNamespace(brand="other", flags=4096, alternativeExperience=256) + preferences = VehicleStartupPreferences(True) + self.assertIs(preferences.prepare(cp), cp) + preferences.finalize(cp) + self.assertEqual((cp.flags, cp.alternativeExperience), (4096, 256)) diff --git a/openpilot/starpilot/tests/test_vehicle_state_context.py b/openpilot/starpilot/tests/test_vehicle_state_context.py new file mode 100644 index 0000000000..68784e1637 --- /dev/null +++ b/openpilot/starpilot/tests/test_vehicle_state_context.py @@ -0,0 +1,98 @@ +"""Synthetic envelopes exercise the actual Card bridge, not native admission.""" +import ast +import inspect +import textwrap +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from openpilot.selfdrive.car import card +from openpilot.starpilot.vehicle_startup import VehicleStartupOwner + + +class Envelope(dict): + def __init__(self): + super().__init__(carControl=SimpleNamespace(enabled=1), pandaStates=[object(), object()]) + self.seen = dict.fromkeys(self, True) + self.valid = dict.fromkeys(self, True) + self.alive = dict.fromkeys(self, True) + self.logMonoTime = {'carControl': 9_950_000_000, 'pandaStates': 29_950_000_000} + self.recv_time = dict.fromkeys(self, 9.95) + + +class TestVehicleStateContext(unittest.TestCase): + def fixture(self): + calls = [] + holder = VehicleStartupOwner() + holder.owner = SimpleNamespace(after_state=lambda **context: calls.append(context)) + instance = SimpleNamespace(vehicle_startup=holder, sm=Envelope(), CI=object()) + return instance, calls + + def invoke(self, instance, clocks=(10_000_000_000, 30_000_000_000)): + state = object() + with patch.object(card, 'clock_pair_ns', return_value=clocks): + card.Car.update_vehicle_state_context(instance, state) + return state + + def test_separated_clocks_and_exact_owner_forwarding(self): + instance, calls = self.fixture() + state = self.invoke(instance) + self.assertEqual(len(calls), 1) + result = calls[0] + self.assertIs(result['state'], state) + self.assertIs(result['ci'], instance.CI) + self.assertIs(result['pandas'], instance.sm['pandaStates']) + self.assertIs(result['control_enabled'], True) + self.assertIs(result['control_current'], True) + self.assertIs(result['panda_current'], True) + self.assertEqual(result['now_ns'], 30_000_000_000) + self.assertEqual(result['panda_log_ns'], 29_950_000_000) + self.assertEqual(result['panda_recv_ns'], 29_950_000_000) + + def test_stale_future_absent_and_invalid_evidence(self): + for service, age in (('carControl', 150_000_001), ('pandaStates', 300_000_001)): + for evidence in ('producer_stale', 'producer_future', 'producer_absent', + 'receipt_stale', 'receipt_future', 'receipt_absent', 'seen', 'valid', 'alive'): + with self.subTest(service=service, evidence=evidence): + instance, calls = self.fixture() + producer_now = 10_000_000_000 if service == 'carControl' else 30_000_000_000 + if evidence.startswith('producer'): + instance.sm.logMonoTime[service] = {'producer_stale': producer_now - age, + 'producer_future': producer_now + 1, 'producer_absent': 0}[evidence] + elif evidence.startswith('receipt'): + instance.sm.recv_time[service] = {'receipt_stale': (10_000_000_000 - age) / 1e9, + 'receipt_future': 10.001, 'receipt_absent': 0}[evidence] + else: + getattr(instance.sm, evidence)[service] = False + self.invoke(instance) + result = calls[0] + self.assertIs(result['control_current' if service == 'carControl' else 'panda_current'], False) + if service == 'pandaStates': + self.assertEqual(result['panda_recv_ns'], 0) + instance, calls = self.fixture() + self.invoke(instance, None) + self.assertIs(calls[0]['control_current'], False) + self.assertIs(calls[0]['panda_current'], False) + self.assertEqual(calls[0]['now_ns'], 0) + + def test_no_holder_or_owner_performs_no_clock_or_envelope_work(self): + for instance in (SimpleNamespace(), SimpleNamespace(vehicle_startup=VehicleStartupOwner())): + with patch.object(card, 'clock_pair_ns', side_effect=AssertionError('unexpected clock read')): + card.Car.update_vehicle_state_context(instance, object()) + VehicleStartupOwner().after_state(unused=object()) + + def test_state_update_source_order_obligation(self): + # The full Card fixture requires transport/hardware. Bind the actual caller + # order explicitly; envelope tests above execute the imported bridge. + tree = ast.parse(textwrap.dedent(inspect.getsource(card.Car.state_update))) + calls = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call): + calls.append((node.lineno, ast.unparse(node.func))) + def position(name): + return min(line for line, function in calls if function == name) + self.assertLess(position('self.CI.update'), position('self.sm.update')) + self.assertLess(position('self.sm.update'), position('self.update_vehicle_state_context')) + self.assertLess(position('self.update_vehicle_state_context'), position('self.v_cruise_helper.update_v_cruise')) + step = inspect.getsource(card.Car.step) + self.assertLess(step.index('self.state_update()'), step.index('self.state_publish(')) diff --git a/openpilot/starpilot/vehicle_preferences.py b/openpilot/starpilot/vehicle_preferences.py new file mode 100644 index 0000000000..f06680026c --- /dev/null +++ b/openpilot/starpilot/vehicle_preferences.py @@ -0,0 +1,203 @@ +from dataclasses import dataclass + +from opendbc.car.gm.startup_preferences import prepare_disable_longitudinal +from opendbc.car.toyota.interface import apply_toyota_auto_hold +from opendbc.car.toyota.values import ToyotaFlags +from openpilot.starpilot.saved_source import read_saved + + +def bolt_disable_supported(cp) -> bool: + from opendbc.car.gm.values import is_bolt_cc_profile, is_bolt_pedal_profile, is_bolt_pedal_stock_denied + identities = ("CHEVROLET_BOLT_CC_2017", "CHEVROLET_BOLT_CC_2018_2021", "CHEVROLET_BOLT_CC_2022_2023", + "CHEVROLET_BOLT_ACC_2022_2023_PEDAL") + return cp.carFingerprint in identities and (is_bolt_cc_profile(cp) or is_bolt_pedal_profile(cp) or + is_bolt_pedal_profile(cp, stock_only=True) or is_bolt_pedal_stock_denied(cp)) + + +@dataclass(frozen=True) +class VehicleStartupPreferences: + toyota_auto_hold: bool = False + turn_assist: bool = False + gm_long_pitch: bool = True + disable_bolt_long: bool = False + honda_bosch_a_radar: bool = True + tesla_preap_stock: bool = True + volt_sng: bool = False + gm_auto_hold: bool = False + volt_one_pedal: bool = False + tesla_screen: bool = False + tesla_screen_brake: bool = False + gm_camera_pedal: bool = True + toyota_filter: bool = True + gm_longitudinal_tune: int = 0 + ford_lateral_mode: int = 0 + toyota_lock_doors: bool = False + toyota_unlock_doors: bool = False + toyota_radar_mode: int = 0 + + @classmethod + def read(cls, params, *, enabled: bool): + from openpilot.starpilot.car.ford.lateral_preferences import selected as ford_mode + from openpilot.starpilot.car.toyota.radar_preferences import selected as toyota_radar_mode + from openpilot.starpilot.car.gm.tune_preferences import selected_tune + from openpilot.starpilot.car.tesla.preap_preferences import stock_configuration + preap_stock = stock_configuration(params) + try: + raw_radar, radar_readable = read_saved(params, "HondaBoschARadar", 8) + honda_radar = radar_readable and raw_radar in (None, b"1") + except (OSError, TypeError, ValueError): + honda_radar = False + try: + raw, readable = read_saved(params, "DisableOpenpilotLongitudinal", 8) + disable_bolt = not readable or raw not in (None, b"0") + except (OSError, TypeError, ValueError): + disable_bolt = True + try: + requested = read_saved(params, "ToyotaAutoHold", 8) == (b"1", True) + safe, readable = read_saved(params, "SafeMode", 8) + toyota = bool(enabled and requested and readable and safe in (None, b"0")) + volt_sng = bool(enabled and read_saved(params, "VoltSNG", 8) == (b"1", True) and + readable and safe in (None, b"0")) + gm_auto_hold = bool(enabled and not disable_bolt and read_saved(params, "GMAutoHold", 8) == (b"1", True) and + readable and safe in (None, b"0")) + volt_one_pedal = bool(enabled and not disable_bolt and read_saved(params, "VoltOnePedalMode", 8) == (b"1", True) and + readable and safe in (None, b"0")) + except (OSError, TypeError, ValueError): + return cls(disable_bolt_long=disable_bolt, honda_bosch_a_radar=honda_radar, tesla_preap_stock=preap_stock, + gm_camera_pedal=False, toyota_filter=False) + try: + pitch, pitch_readable = read_saved(params, "LongPitch", 8) + pitch_enabled = not (enabled and pitch_readable and pitch == b"0" and readable and safe in (None, b"0")) + except (OSError, TypeError, ValueError): + pitch_enabled = True + try: + assist_raw, assist_readable = read_saved(params, "TurnAssist", 8) + assist = assist_readable and assist_raw in (None, b"1") + except (OSError, TypeError, ValueError): + assist = False + screen = bool(enabled and readable and safe in (None, b"0") and + read_saved(params, 'AlwaysOnLateral', 8) == (b'1', True) and + read_saved(params, 'TeslaAOLScreenTap', 8) == (b'1', True)) + screen_brake = read_saved(params, 'TeslaAOLDisengageOnBrake', 8) == (b'1', True) + return cls(toyota_radar_mode=toyota_radar_mode(params, enabled=enabled), + ford_lateral_mode=ford_mode(params, enabled=enabled), tesla_screen=screen, + tesla_screen_brake=screen_brake, toyota_auto_hold=toyota, + gm_longitudinal_tune=selected_tune(params, enabled=enabled), + toyota_lock_doors=bool(enabled and readable and safe in (None, b"0") and + read_saved(params, "LockDoors", 8) == (b"1", True)), + toyota_unlock_doors=bool(enabled and readable and safe in (None, b"0") and + read_saved(params, "UnlockDoors", 8) == (b"1", True)), + volt_sng=volt_sng, gm_auto_hold=gm_auto_hold, volt_one_pedal=volt_one_pedal, + gm_camera_pedal=bool(enabled and not disable_bolt and readable and safe in (None, b"0")), + toyota_filter=bool(enabled and not disable_bolt and readable and safe in (None, b"0")), + turn_assist=bool(enabled and assist and readable and safe in (None, b"0")), + gm_long_pitch=pitch_enabled, disable_bolt_long=disable_bolt, honda_bosch_a_radar=honda_radar, tesla_preap_stock=preap_stock) + + def _prepare_honda_radar(self, cp) -> None: + if cp.brand != "honda": + return + from opendbc.car.honda.bosch_a_radar import VERIFIED_BOSCH_A_CARS + from opendbc.car.honda.values import HondaFlags + if (cp.brand == "honda" and cp.carFingerprint in VERIFIED_BOSCH_A_CARS and cp.flags & HondaFlags.BOSCH and + not cp.flags & (HondaFlags.BOSCH_RADARLESS | HondaFlags.BOSCH_CANFD | HondaFlags.BOSCH_ALT_RADAR)): + cp.radarUnavailable = not self.honda_bosch_a_radar + + def _prepare_bolt(self, cp, fingerprints=None) -> None: + if self.disable_bolt_long and bolt_disable_supported(cp): + from opendbc.car.gm.values import (CAR, GMSafetyFlags, is_bolt_pedal_profile, is_bolt_pedal_stock_denied, + is_bolt_pedal_removed_profile, BOLT_PEDAL_REMOVED_STOCK_WORDS) + from opendbc.car.structs import CarParams + pedal = is_bolt_pedal_profile(cp) or is_bolt_pedal_profile(cp, stock_only=True) or is_bolt_pedal_stock_denied(cp) + stock_qualified = is_bolt_pedal_profile(cp, stock_only=True) + removed_pedal = is_bolt_pedal_removed_profile(cp) or is_bolt_pedal_removed_profile(cp, stock_only=True) + denied = is_bolt_pedal_stock_denied(cp) + if removed_pedal: + cp.safetyConfigs[0].safetyParam = BOLT_PEDAL_REMOVED_STOCK_WORDS[cp.carFingerprint] + cp.openpilotLongitudinalControl = False + cp.pcmCruise = True + cp.autoResumeSng = False + return + cp.openpilotLongitudinalControl = False + if pedal: + cp.pcmCruise = True + if cp.carFingerprint == CAR.CHEVROLET_BOLT_ACC_2022_2023_PEDAL: + if stock_qualified or denied: + return + camera = fingerprints.get(2, {}) if isinstance(fingerprints, dict) else {} + length = camera.get(0x180) if isinstance(camera, dict) else None + if type(length) is int and length == 4: + cp.safetyConfigs[0].safetyParam = int(GMSafetyFlags.HW_CAM | GMSafetyFlags.EV) + else: + safety = CarParams.SafetyConfig() + safety.safetyModel = CarParams.SafetyModel.noOutput + cp.safetyConfigs = [safety] + cp.passive = True + cp.dashcamOnly = True + + def _prepare_camera_pedal(self, cp) -> None: + from opendbc.car.gm.values import camera_acc_pedal_profile, volt_cc_pedal_profile + if not self.gm_camera_pedal and (camera_acc_pedal_profile(cp) is not None or volt_cc_pedal_profile(cp) is not None): + prepare_disable_longitudinal(cp, True) + + def prepare(self, cp, *, fingerprints=None): + from opendbc.car.gm.values import apply_gm_auto_hold, apply_volt_one_pedal + from openpilot.starpilot.car.tesla.preap_preferences import prepare_stock + prepare_stock(cp, self.tesla_preap_stock) + from opendbc.car.tesla.screen_button import apply_screen_button + apply_screen_button(cp, self.tesla_screen, self.tesla_screen_brake) + self._prepare_honda_radar(cp) + if not self.toyota_filter: + from opendbc.car.toyota.prius_longitudinal import prepare_stock as prepare_prius_stock + prepare_prius_stock(cp) + self._prepare_bolt(cp, fingerprints) + prepare_disable_longitudinal(cp, self.disable_bolt_long) + self._prepare_camera_pedal(cp) + apply_gm_auto_hold(cp, self.gm_auto_hold) + apply_volt_one_pedal(cp, self.volt_one_pedal and not self.disable_bolt_long, self.gm_auto_hold) + if cp.brand == "toyota": + apply_toyota_auto_hold(cp, self.toyota_auto_hold) + from opendbc.car.toyota.ars510_preferences import apply_radar_mode + apply_radar_mode(cp, self.toyota_radar_mode) + from opendbc.bluepilot_lateral.hosts.starpilot import select as select_ford_lateral + select_ford_lateral(cp, self.ford_lateral_mode) + return cp + + def finalize(self, cp) -> None: + from opendbc.car.gm.values import apply_gm_auto_hold, is_gm_auto_hold, apply_volt_one_pedal, is_volt_one_pedal + admitted_one_pedal = is_volt_one_pedal(cp) + admitted_hold = is_gm_auto_hold(cp) + from openpilot.starpilot.car.tesla.preap_preferences import prepare_stock + prepare_stock(cp, self.tesla_preap_stock) + from opendbc.car.tesla.screen_button import apply_screen_button + apply_screen_button(cp, self.tesla_screen, self.tesla_screen_brake) + self._prepare_honda_radar(cp) + if not self.toyota_filter: + from opendbc.car.toyota.prius_longitudinal import prepare_stock as prepare_prius_stock + prepare_prius_stock(cp) + self._prepare_bolt(cp) + prepare_disable_longitudinal(cp, self.disable_bolt_long) + self._prepare_camera_pedal(cp) + apply_gm_auto_hold(cp, self.gm_auto_hold and admitted_hold) + apply_volt_one_pedal(cp, self.volt_one_pedal and admitted_one_pedal and not self.disable_bolt_long, self.gm_auto_hold and admitted_hold) + if cp.brand == "toyota": + admitted = bool(cp.flags & ToyotaFlags.AUTO_BRAKE_HOLD) + apply_toyota_auto_hold(cp, self.toyota_auto_hold and admitted) + from opendbc.car.toyota.ars510_preferences import apply_radar_mode + apply_radar_mode(cp, self.toyota_radar_mode) + + def configure_controller(self, ci) -> None: + from opendbc.car.gm.values import (CAR, is_bolt_euv_longitudinal, is_volt_longitudinal, is_gm_auto_hold, + is_volt_one_pedal, camera_acc_pedal_profile, volt_cc_pedal_profile) + cp = ci.CP + if ci.CC is not None and cp.brand == "toyota" and not cp.passive and not cp.dashcamOnly and not cp.notCar: + ci.CC.door_locks.lock = self.toyota_lock_doors + ci.CC.door_locks.unlock = self.toyota_unlock_doors + if ci.CC is not None and is_volt_longitudinal(cp): + ci.CC.volt_sng = self.volt_sng + if ci.CC is not None and cp.brand == "gm": + ci.CC.gm_auto_hold = self.gm_auto_hold and is_gm_auto_hold(cp) + ci.CC.volt_one_pedal = self.volt_one_pedal and is_volt_one_pedal(cp) + pedal = camera_acc_pedal_profile(cp) or volt_cc_pedal_profile(cp) + if ci.CC is not None and (is_bolt_euv_longitudinal(cp) or is_volt_longitudinal(cp) or cp.carFingerprint == CAR.CHEVROLET_SUBURBAN or + pedal is not None and pedal.longitudinal): + ci.CC.long_pitch = self.gm_long_pitch diff --git a/openpilot/starpilot/vehicle_startup.py b/openpilot/starpilot/vehicle_startup.py new file mode 100644 index 0000000000..a716a61009 --- /dev/null +++ b/openpilot/starpilot/vehicle_startup.py @@ -0,0 +1,69 @@ +"""Generic lifetime holder for a car-owned prepublication startup transaction.""" +import threading + + +class VehicleStartupOwner: + def __init__(self, owner=None): + # Explicit-CI callers may supply an already prepared car-owned decision. + self.owner = owner + self.send_lock = threading.Lock() + + def prepare(self, cp, interface, callbacks, *, requested, admission): + factory = getattr(interface, 'startup_owner', None) + if factory is None: + return cp + recv, send = callbacks + def locked_send(frames): + with self.send_lock: + send(frames) + owner = factory(cp, (recv, locked_send), requested=requested) + if owner is None: + return cp + # Register before any ECU request so a partially constructed Card can close. + self.owner = owner + return owner.prepare(admission=admission) + + def configure(self, ci): + required = getattr(type(ci), 'startup_required', None) + if required is not None and required(ci.CP): + matches = getattr(self.owner, 'prepared_for', None) + if matches is None or not matches(ci.CP): + raise RuntimeError('Car interface requires a matching prepared startup owner') + if self.owner is not None: + self.owner.configure(ci) + + def finalize_aol_configuration(self, ci): + finalize = getattr(self.owner, 'finalize_aol_configuration', None) + if finalize is not None: + finalize(ci) + + def seal_publication(self): + if self.owner is not None: + self.owner.seal_publication() + + def check(self): + if self.owner is not None: + self.owner.check() + + def after_state(self, **context): + if self.owner is not None: + self.owner.after_state(**context) + + def consume_cruise_resume(self): + return self.owner.consume_cruise_resume() if self.owner is not None else None + + def before_control(self, *, configured, ci, now_ns, control_current): + return self.owner is None or self.owner.before_control( + configured=configured, sources_current=self.owner.sources_current(ci, now_ns), control_current=control_current) + + def maintain(self, *, configured): + if self.owner is not None: + self.owner.maintain(configured=configured) + + def sent(self, frames, *, valid): + if self.owner is not None: + self.owner.sent(frames, valid=valid) + + def close(self): + if self.owner is not None: + self.owner.close()