From c99cca360d43d83375b68381df005e4e632b4472 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:19:26 -0600 Subject: [PATCH] Qualify queued Ioniq 6 blind-spot warnings --- .github/bsm-qualification.py | 133 ++++++++++++++++++ .github/workflows/ioniq6-bsm-check.yaml | 85 +++++++++++ .../opendbc/car/hyundai/carcontroller.py | 6 +- .../car/hyundai/tests/test_ioniq6_bsm.py | 92 +++++++++++- .../opendbc/safety/modes/hyundai_canfd.h | 9 +- .../safety/tests/test_hyundai_ioniq6_long.py | 77 +++++----- 6 files changed, 356 insertions(+), 46 deletions(-) create mode 100644 .github/bsm-qualification.py create mode 100644 .github/workflows/ioniq6-bsm-check.yaml diff --git a/.github/bsm-qualification.py b/.github/bsm-qualification.py new file mode 100644 index 0000000000..822d1463fe --- /dev/null +++ b/.github/bsm-qualification.py @@ -0,0 +1,133 @@ +from pathlib import Path +import hashlib +import json +import os +import subprocess +import sys +import xml.etree.ElementTree as ET + +BASE = '1d30dc343469e93967d2de3afdc966901009ed5d' +OUT = Path('bsm-results') +OUT.mkdir(exist_ok=True) +HOST = 'opendbc_repo/opendbc/car/hyundai/tests/test_ioniq6_bsm.py' +NATIVE = 'opendbc_repo/opendbc/safety/tests/test_hyundai_ioniq6_long.py' + +def run(*args): + subprocess.run(args, check=True) + +def qualify(): + controller = Path('opendbc_repo/opendbc/car/hyundai/carcontroller.py') + shipping = controller.read_bytes() + try: + controller.write_bytes(subprocess.check_output(['git', 'show', BASE + ':' + str(controller)])) + result = subprocess.run([sys.executable, '-m', 'pytest', '-q', HOST, '-k', + 'requires_a_matching_queued', '--junitxml=bsm-results/baseline.xml']) + cases = list(ET.parse(OUT / 'baseline.xml').getroot().iter('testcase')) + failures = [case for case in cases if case.find('failure') is not None] + assert result.returncode == 1 and failures and not list(ET.parse(OUT / 'baseline.xml').iter('error')) + assert all('UNQUEUED_BLINDSPOT_WARNING' in ''.join(case.find('failure').itertext()) for case in failures) + (OUT / 'baseline.json').write_text(json.dumps({'expectedFailingCases': len(failures), 'confirmed': True})) + finally: + controller.write_bytes(shipping) + assert controller.read_bytes() == shipping + suites = [HOST, *map(str, sorted(Path('opendbc_repo/opendbc/car/hyundai/tests').glob('test_ioniq6_*.py'))), + 'openpilot/starpilot/lateral/tests/test_auto_lane_change.py', + 'openpilot/starpilot/lateral/tests/test_lane_change_defaults.py', + 'openpilot/starpilot/lateral/tests/test_lane_change_floor.py'] + run(sys.executable, '-m', 'pytest', '-q', '--junitxml=bsm-results/host.xml', *dict.fromkeys(suites)) + from tools.ci.run_vehicle_tests import build_safety_library + native = [NATIVE, + 'opendbc_repo/opendbc/safety/tests/test_hyundai_canfd.py', + 'opendbc_repo/opendbc/safety/tests/test_hyundai_canfd_carnival.py', + 'opendbc_repo/opendbc/safety/tests/test_hyundai_canfd_torque_envelope.py', + 'opendbc_repo/opendbc/safety/tests/test_hyundai_ioniq6_stock_aol.py'] + for release in (False, True): + variant = 'release' if release else 'debug' + provenance = build_safety_library(release, OUT) + (OUT / f'native-{variant}.json').write_text(json.dumps(provenance, indent=2)) + # A fresh interpreter avoids mixing the two C global-state libraries. + loader = ('from opendbc.safety.tests.libsafety import libsafety_py; ' + f'libsafety_py.load({str((OUT / ("libsafety-" + variant + ".so")).resolve())!r}); ' + 'import pytest,sys; sys.exit(pytest.main(sys.argv[1:]))') + run(sys.executable, '-c', loader, '-q', f'--junitxml=bsm-results/native-{variant}.xml', *native) + paths = [str(controller), HOST, NATIVE] + run('.venv/bin/ruff', 'check', *paths) + run('.venv/bin/ty', 'check', *paths) + (OUT / 'source.json').write_text(json.dumps({str(p): hashlib.sha256(p.read_bytes()).hexdigest() + for p in [controller, Path(HOST), Path(NATIVE), Path('opendbc_repo/opendbc/safety/modes/hyundai_canfd.h')]}, indent=2)) + +def firmware(): + variant = os.environ['BSM_VARIANT'] + names = subprocess.check_output(['git', 'ls-tree', '-r', '--name-only', 'HEAD', 'panda/board/obj'], text=True).splitlines() + targets = [name for name in names if name.endswith(('.bin.signed', '.bin'))] + targets += ['panda/board/obj/version', 'panda/board/obj/gitversion.h', 'panda/board/obj/cert.h'] + assert sum(name.endswith('.bin.signed') for name in targets) == 17 + with (OUT / 'firmware-build.log').open('w') as log: + result = subprocess.run(['.venv/bin/scons', '-j2', '--verbose', *targets], stdout=log, stderr=subprocess.STDOUT) + print((OUT / 'firmware-build.log').read_text()[-18000:]) + assert result.returncode == 0 + run(sys.executable, 'tools/ci/record_firmware.py', '--variant', variant, '--build-log', + str(OUT / 'firmware-build.log'), '--output', str(OUT / 'firmware-provenance.json')) + import base64 + import zlib + key = base64.b64decode(Path('panda/board/crypto/certs/debug.pub').read_bytes().split()[1]) + fields = [] + for _ in range(3): + size = int.from_bytes(key[:4], 'big') + fields.append(key[4:4 + size]) + key = key[4 + size:] + exponent, modulus = map(lambda value: int.from_bytes(value, 'big'), fields[1:]) + checks = {} + for name in targets: + data = Path(name).read_bytes() + if name.endswith('.bin.signed'): + payload, signature = data[:-128], data[-128:] + expected = b'\x00\x01' + b'\xff' * 0x69 + b'\x00' + hashlib.sha1(payload[4:]).digest() + assert pow(int.from_bytes(signature, 'big'), exponent, modulus).to_bytes(128, 'big') == expected + assert int.from_bytes(payload[:4], 'little') == len(payload) + assert int.from_bytes(payload[-12:-8], 'little') == zlib.crc32(payload[:-12]) + assert payload[-8:-4] == b'VERS' + elf = name.removesuffix('.bin.signed') + '/main.elf' + symbols = subprocess.check_output(['arm-none-eabi-nm', elf], text=True) + if variant == 'debug' and not name.endswith('/body_h7.bin.signed'): + assert 'hyundai_canfd_ioniq6_bsm_wheel_valid' in symbols + if '/bootstub.' in name: + project = Path(name).name.removeprefix('bootstub.').removesuffix('.bin') + symbols = subprocess.check_output(['arm-none-eabi-nm', f'panda/board/obj/{project}/bootstub.elf'], text=True) + assert ' RSA_verify' in symbols + checks[name] = {'sha256': hashlib.sha256(data).hexdigest(), 'bytes': len(data), 'mode': Path(name).stat().st_mode & 511} + (OUT / 'all-firmware.json').write_text(json.dumps({'variant': variant, 'artifacts': checks}, indent=2)) + if variant == 'debug': + import tarfile + with tarfile.open(OUT / 'shipping-firmware.tar', 'w') as archive: + for name in targets: + archive.add(name, arcname=name) + +def helper(): + import ctypes + import struct + target = Path('panda/tests/libpanda/libpanda.so') + original = OUT / 'parent-libpanda.so' + original.write_bytes(target.read_bytes()) + run('cc', '-fPIC', '-nostdlib', '-fno-builtin', '-std=gnu11', '-Wfatal-errors', '-Wno-pointer-to-int-cast', + '-Ipanda/tests/libpanda', '-Ipanda', '-Ipanda/board', '-Iopendbc_repo', '-c', + 'panda/tests/libpanda/panda.c', '-o', str(OUT / 'panda.os')) + run('cc', '-shared', str(OUT / 'panda.os'), '-o', str(target)) + for path in (original, target): + assert struct.unpack_from('- + .venv/bin/scons -j2 + openpilot/starpilot/ui/native/_raylib_extra.so + openpilot/common/libparams_c.so + msgq_repo/msgq/_ipc.so + msgq_repo/msgq/visionipc/_visionipc.so + opendbc_repo/opendbc/can/_raw_value.so + opendbc_repo/opendbc/dbc + openpilot/selfdrive/controls/lib/longitudinal_mpc_lib/c_generated_code/acados_ocp_solver_pyx.so + openpilot/selfdrive/locationd/models/generated/libcar.so + - run: .venv/bin/python .github/bsm-qualification.py qualify + - uses: actions/upload-artifact@v4 + if: always() + with: + name: bsm-host-native + path: bsm-results/ + if-no-files-found: error + firmware: + strategy: + fail-fast: false + matrix: + variant: [debug, release] + runs-on: ubuntu-24.04 + timeout-minutes: 30 + env: + PYTHONPATH: ${{ github.workspace }} + BSM_VARIANT: ${{ matrix.variant }} + steps: + - uses: actions/checkout@v7 + - run: ./tools/op.sh setup + - name: Build and verify all firmware variants + shell: bash + run: | + if [[ "$BSM_VARIANT" == release ]]; then + export RELEASE=1 + export CERT="$PWD/panda/board/crypto/certs/debug" + else + unset RELEASE + fi + .venv/bin/python .github/bsm-qualification.py firmware + - uses: actions/upload-artifact@v4 + if: always() + with: + name: bsm-firmware-${{ matrix.variant }} + path: bsm-results/ + if-no-files-found: error + arm64-helper: + runs-on: ubuntu-24.04-arm + timeout-minutes: 15 + steps: + - uses: actions/checkout@v7 + - run: python3 .github/bsm-qualification.py helper + - uses: actions/upload-artifact@v4 + if: always() + with: + name: bsm-arm64-helper + path: bsm-results/ + if-no-files-found: error diff --git a/opendbc_repo/opendbc/car/hyundai/carcontroller.py b/opendbc_repo/opendbc/car/hyundai/carcontroller.py index 1251902518..abfba2aba2 100644 --- a/opendbc_repo/opendbc/car/hyundai/carcontroller.py +++ b/opendbc_repo/opendbc/car/hyundai/carcontroller.py @@ -597,8 +597,10 @@ class CarController(CarControllerBase): can_now_ns = getattr(CS, "ioniq6_bsm_now_ns", 0) status = (sources.status(can_now_ns) if sources is not None and can_now_ns > self.ioniq6_bsm_last_can_ns else None) if status is not None: - # Neutralize only the injected dashboard indication below 5 mph. - if CS.out.vEgoRaw < 5 * CV.MPH_TO_MS: + left_queued = CC.leftBlinker and not CC.rightBlinker + right_queued = CC.rightBlinker and not CC.leftBlinker + if (CS.out.vEgoRaw < 5 * CV.MPH_TO_MS or + status.left == 2 and not left_queued or status.right == 2 and not right_queued): status = BlindspotStatus(0, 0) can_sends.extend(hyundaicanfd.create_ioniq6_blindspot_status(self.ioniq6_bsm_counter, status)) self.ioniq6_bsm_counter = (self.ioniq6_bsm_counter + 1) & 0xff diff --git a/opendbc_repo/opendbc/car/hyundai/tests/test_ioniq6_bsm.py b/opendbc_repo/opendbc/car/hyundai/tests/test_ioniq6_bsm.py index 4c4fa0f5cf..eca39bf984 100644 --- a/opendbc_repo/opendbc/car/hyundai/tests/test_ioniq6_bsm.py +++ b/opendbc_repo/opendbc/car/hyundai/tests/test_ioniq6_bsm.py @@ -168,10 +168,12 @@ class TestIoniq6BlindspotStatus(unittest.TestCase): lamp_source_ns=t, left_lamp=left_lamp, right_lamp=right_lamp) ci.CS.ioniq6_bsm_sources = sources source_status = sources.status(t) + assert source_status is not None counter = controller.ioniq6_bsm_counter _, frames = controller.update(control.as_reader(), ci.CS, t) status_frames = [(addr, payload, bus) for addr, payload, bus in frames if addr in (0x1BA, 0x1E5)] - expected = BlindspotStatus(0, 0) if speed < threshold else source_status + expected = (BlindspotStatus(0, 0) if speed < threshold or 2 in (source_status.left, source_status.right) + else source_status) self.assertEqual(status_frames, create_ioniq6_blindspot_status(counter, expected)) self.assertEqual(sources.status(t), source_status) for addr, payload, _ in status_frames: @@ -182,3 +184,91 @@ class TestIoniq6BlindspotStatus(unittest.TestCase): _, skipped = controller.update(control.as_reader(), ci.CS, t + i * 10_000_000) self.assertFalse(any(addr in (0x1BA, 0x1E5) for addr, _, _ in skipped)) ci.CS.out = out + + + def test_controller_requires_a_matching_queued_lane_change_for_escalation(self): + cp = long_params() + ci = CarInterface(cp) + controller = CarController(DBC[cp.carFingerprint], cp) + packer = CANPacker(DBC[cp.carFingerprint][Bus.pt]) + control = structs.CarControl() + controller.ioniq6_bsm_counter = 250 + for physical_left in (True, False): + for request in ('matching', 'none', 'opposite', 'hazards', 'matching', 'none'): + with self.subTest(left=physical_left, request=request): + t = 10_000_000_000 + controller.frame * 10_000_000 + ci.update([(t, [packer.make_can_msg('BLINDSPOTS_FRONT_CORNER_2', 1, + {'SIDE_DETECT_STATE': 0x10 if physical_left else 0x08}), + packer.make_can_msg('BLINKERS', 1, {})])]) + ci.CS.out.vEgoRaw = 10.0 + sources = Ioniq6BlindspotSources() + sources.observe(corner_source_ns=t, corner_state=0x10 if physical_left else 0x08, + lamp_source_ns=t, left_lamp=physical_left, right_lamp=not physical_left) + ci.CS.ioniq6_bsm_sources = sources + control.leftBlinker = request == 'hazards' or request == ('matching' if physical_left else 'opposite') + control.rightBlinker = request == 'hazards' or request == ('opposite' if physical_left else 'matching') + counter = controller.ioniq6_bsm_counter + _, frames = controller.update(control.as_reader(), ci.CS, t) + actual = [(addr, payload, bus) for addr, payload, bus in frames if addr in (0x1BA, 0x1E5)] + source_status = sources.status(t) + assert source_status is not None + expected = source_status if request == 'matching' else BlindspotStatus(0, 0) + self.assertEqual(actual, create_ioniq6_blindspot_status(counter, expected), 'UNQUEUED_BLINDSPOT_WARNING') + self.assertEqual(sources.status(t), BlindspotStatus(2, 0) if physical_left else BlindspotStatus(0, 2)) + for offset in range(1, 5): + _, skipped = controller.update(control.as_reader(), ci.CS, t + offset * 10_000_000) + self.assertFalse(any(addr in (0x1BA, 0x1E5) for addr, _, _ in skipped)) + + def test_actual_lane_change_request_completion_and_turn_with_held_stalk(self): + from openpilot.cereal import log + from openpilot.selfdrive.controls.lib.desire_helper import DesireHelper + from openpilot.starpilot.lateral.lane_change_preferences import LaneChangePolicy + + for left in (True, False): + cp = long_params() + ci = CarInterface(cp) + controller = CarController(DBC[cp.carFingerprint], cp) + state = structs.CarState(vEgo=20.0, leftBlinker=left, rightBlinker=not left) + helper = DesireHelper(LaneChangePolicy(auto_lane_change=True)) + control = structs.CarControl() + + def check(expected): + t = 10_000_000_000 + controller.frame * 10_000_000 + ci.CS.ioniq6_bsm_now_ns = t + ci.CS.out.vEgoRaw = 20.0 + sources = Ioniq6BlindspotSources() + sources.observe(corner_source_ns=t, corner_state=0x10 if left else 0x08, + lamp_source_ns=t, left_lamp=left, right_lamp=not left) + ci.CS.ioniq6_bsm_sources = sources + queued = helper.lane_change_state != log.LaneChangeState.off + control.leftBlinker = queued and helper.lane_change_direction == log.LaneChangeDirection.left + control.rightBlinker = queued and helper.lane_change_direction == log.LaneChangeDirection.right + counter = controller.ioniq6_bsm_counter + _, frames = controller.update(control.as_reader(), ci.CS, t) + self.assertEqual([frame for frame in frames if frame[0] in (0x1BA, 0x1E5)], + create_ioniq6_blindspot_status(counter, expected)) + for offset in range(1, 5): + controller.update(control.as_reader(), ci.CS, t + offset * 10_000_000) + + # Missing automatic-lane evidence does not cancel a manual pending request. + helper.update(state, True, 1.0, auto_evidence=False, engaged=True) + helper.update(state, True, 1.0, auto_evidence=False, engaged=True) + self.assertEqual(helper.auto_status, 'laneUnavailable') + self.assertEqual(helper.lane_change_state, log.LaneChangeState.preLaneChange) + check(BlindspotStatus(2, 0) if left else BlindspotStatus(0, 2)) + state.steeringPressed = True + state.steeringTorque = 1.0 if left else -1.0 + helper.update(state, True, 1.0, engaged=True) + self.assertEqual(helper.lane_change_state, log.LaneChangeState.laneChangeStarting) + check(BlindspotStatus(2, 0) if left else BlindspotStatus(0, 2)) + state.steeringPressed = False + for _ in range(12): + helper.update(state, True, 0.0, engaged=True) + self.assertEqual(helper.lane_change_state, log.LaneChangeState.off) + check(BlindspotStatus(0, 0)) + helper.update(state, True, 0.0, engaged=True) + check(BlindspotStatus(0, 0)) + helper = DesireHelper() + helper.update(state, True, 1.0, navigation_turn=True) + self.assertEqual(helper.lane_change_state, log.LaneChangeState.off) + check(BlindspotStatus(0, 0)) diff --git a/opendbc_repo/opendbc/safety/modes/hyundai_canfd.h b/opendbc_repo/opendbc/safety/modes/hyundai_canfd.h index 73a03d4c65..af8706f8d3 100644 --- a/opendbc_repo/opendbc/safety/modes/hyundai_canfd.h +++ b/opendbc_repo/opendbc/safety/modes/hyundai_canfd.h @@ -225,7 +225,7 @@ static bool hyundai_canfd_gv60_heartbeat_seen = false; static uint8_t hyundai_canfd_gv60_heartbeat_counter = 0U; static bool hyundai_canfd_ioniq6_heartbeat_seen = false; static uint8_t hyundai_canfd_ioniq6_heartbeat_counter = 0U; -static bool hyundai_canfd_ioniq6_bsm_low_speed = false; +static bool hyundai_canfd_ioniq6_bsm_wheel_valid = false; static bool hyundai_canfd_ioniq6_bsm_wheel_seen = false; static uint8_t hyundai_canfd_ioniq6_bsm_wheel_counter = 0U; static uint32_t hyundai_canfd_ioniq6_bsm_wheel_ts = 0U; @@ -763,11 +763,10 @@ static void hyundai_canfd_rx_hook(const CANPacket_t *msg) { // average of all 4 wheel speeds. Conversion: raw * 0.03125 / 3.6 = m/s UPDATE_VEHICLE_SPEED((fr + rr + rl + fl) / 4.0 * 0.03125 * KPH_TO_MS); if (hyundai_canfd_ioniq6_long) { - // Mean wheel speed < 5 mph: raw sum < 2.2352 * 4 * 3.6 / 0.03125. // This hook runs only for valid required RX; TX also checks its freshness. const bool fresh_counter = !hyundai_canfd_ioniq6_bsm_wheel_seen || (msg->data[2] == (uint8_t)(hyundai_canfd_ioniq6_bsm_wheel_counter + 1U)); - hyundai_canfd_ioniq6_bsm_low_speed = fresh_counter && ((fr + rr + rl + fl) < 1030U); + hyundai_canfd_ioniq6_bsm_wheel_valid = fresh_counter; if (fresh_counter) { hyundai_canfd_ioniq6_bsm_wheel_ts = microsecond_timer_get(); } @@ -1037,7 +1036,7 @@ static bool hyundai_canfd_tx_hook_valid(const CANPacket_t *msg) { body[7] = 0x80U | (SAFETY_MAX(left, right) << 3U); body[16] = left | (right << 2U); bool physical_body = true; - bool neutral_body = hyundai_canfd_ioniq6_bsm_low_speed && + bool neutral_body = hyundai_canfd_ioniq6_bsm_wheel_valid && (safety_get_ts_elapsed(now, hyundai_canfd_ioniq6_bsm_wheel_ts) <= 100000U); for (int i = 3; i < 24; i++) { physical_body &= msg->data[i] == body[i]; @@ -1314,7 +1313,7 @@ static void hyundai_canfd_reset_ioniq6_state(void) { hyundai_canfd_gv60_heartbeat_counter = 0U; hyundai_canfd_ioniq6_heartbeat_seen = false; hyundai_canfd_ioniq6_heartbeat_counter = 0U; - hyundai_canfd_ioniq6_bsm_low_speed = false; + hyundai_canfd_ioniq6_bsm_wheel_valid = false; hyundai_canfd_ioniq6_bsm_wheel_seen = false; hyundai_canfd_ioniq6_bsm_wheel_counter = 0U; hyundai_canfd_ioniq6_bsm_wheel_ts = 0U; diff --git a/opendbc_repo/opendbc/safety/tests/test_hyundai_ioniq6_long.py b/opendbc_repo/opendbc/safety/tests/test_hyundai_ioniq6_long.py index 776e580504..2bf8d52dab 100644 --- a/opendbc_repo/opendbc/safety/tests/test_hyundai_ioniq6_long.py +++ b/opendbc_repo/opendbc/safety/tests/test_hyundai_ioniq6_long.py @@ -351,11 +351,11 @@ class TestHyundaiIoniq6Long(unittest.TestCase): ("WHL_SpdFLVal", "WHL_SpdFRVal", "WHL_SpdRLVal", "WHL_SpdRRVal"), raw_speeds, strict=True)} return packer.make_can_msg("WHEEL_SPEEDS", 1, {**values, "COUNTER": counter}) - def test_bsm_neutral_only_below_physical_speed_boundary(self): + def test_bsm_neutral_advisory_requires_fresh_wheels_at_all_speeds(self): if self.release: self.skipTest("RELEASE denies the Ioniq 6 LONG profiles") - # 5 mph is between representable sums 1029 and 1030; test unequal wheels too. - samples = ((0, 0, 0, 0), (257, 257, 257, 258), (257, 257, 258, 258), (258, 258, 258, 258)) + # Keep the old speed boundary and highway speeds in the advisory matrix. + samples = ((0, 0, 0, 0), (257, 257, 257, 258), (257, 257, 258, 258), (258, 258, 258, 258), (4000, 4000, 4000, 4000)) for raw in (*RAW_LONG.values(), *RAW_AOL.values()): _, cp = params("lkas_alt" if raw & 0x80 else "lkas") packer = CANPacker(DBC[cp.carFingerprint][Bus.pt]) @@ -365,9 +365,8 @@ class TestHyundaiIoniq6Long(unittest.TestCase): with self.subTest(profile=hex(raw), speeds=speeds): self.assertTrue(self.safety.safety_rx_hook(self.packet(self.bsm_wheel_frame(packer, speeds, counter)))) rear, front = create_ioniq6_blindspot_status(counter, BlindspotStatus(0, 0)) - permitted = sum(speeds) < 1030 - self.assertEqual(self.safety.safety_tx_hook(self.packet(rear)), permitted) - self.assertEqual(self.safety.safety_tx_hook(self.packet(front)), permitted) + self.assertTrue(self.safety.safety_tx_hook(self.packet(rear))) + self.assertTrue(self.safety.safety_tx_hook(self.packet(front))) # Original physical indications remain valid, independently of speed. for frame in create_ioniq6_blindspot_status(counter + 30, BlindspotStatus(2, 2)): self.assertTrue(self.safety.safety_tx_hook(self.packet(frame))) @@ -429,10 +428,10 @@ class TestHyundaiIoniq6Long(unittest.TestCase): self.assertTrue(self.safety.safety_tx_hook(self.packet(rear))) self.safety.set_timer(1_200_002) self.assertFalse(self.safety.safety_tx_hook(self.packet(front))) - # A low-speed packet on the wrong route or with the wrong length cannot - # replace a valid above-threshold physical wheel observation. + # Wrong-route and short packets cannot refresh expired wheel evidence. other_sources(5, 1_210_000) self.assertTrue(self.safety.safety_rx_hook(self.packet(self.bsm_wheel_frame(packer, (300, 300, 300, 300), 5)))) + other_sources(6, 1_310_001) low = self.bsm_wheel_frame(packer, (0, 0, 0, 0), 6) for invalid in ((low[0], low[1], 0), (low[0], low[1][:8], 1)): self.safety.safety_rx_hook(self.packet(invalid)) @@ -478,36 +477,38 @@ class TestHyundaiIoniq6Long(unittest.TestCase): for topology in ("lkas", "lkas_alt"): for aol in (False, True): for raw_speeds in ((0, 0, 0, 0), (257, 257, 257, 258), (257, 257, 258, 258), (300, 300, 300, 300)): - with self.subTest(topology=topology, aol=aol, speeds=raw_speeds): - _, cp = params(topology) - raw = (RAW_AOL if aol else RAW_LONG)[topology] - cp.safetyConfigs[-1].safetyParam = raw - ci = CarInterface(cp) - controller = CarController(DBC[cp.carFingerprint], cp) - packer = CANPacker(DBC[cp.carFingerprint][Bus.pt]) - bus = CanBus(cp) - inputs = [packer.make_can_msg(name, bus.ECAN, values) for name, values in ( - ("ACCELERATOR", {"GEAR": 5, "COUNTER": 1}), ("TCS", {"COUNTER": 1}), - ("WHEEL_SPEEDS", {"COUNTER": 1}), ("MDPS", {"COUNTER": 1}), - ("CRUISE_BUTTONS", {"COUNTER": 1}), - ("BLINDSPOTS_FRONT_CORNER_2", {"COUNTER": 1, "SIDE_DETECT_STATE": 0x18}), - ("BLINKERS", {"LEFT_LAMP": 1}), - )] - inputs = [frame for frame in inputs if frame[0] != 0xA0] - inputs.append(self.bsm_wheel_frame(packer, raw_speeds, 1)) - inputs.append(packer.make_can_msg("CAM_0x362" if topology == "lkas_alt" else "CAM_0x2a4", bus.CAM, {})) - ci.update([(1_000_000_000, inputs)]) - self.mode(raw) - for frame in inputs: - if frame[2] == bus.ECAN: - self.assertTrue(self.safety.safety_rx_hook(self.packet(frame)), hex(frame[0])) - control = structs.CarControl() - _, sent = controller.update(control.as_reader(), ci.CS, 1_000_000_000) - pair = [frame for frame in sent if frame[0] in (0x1BA, 0x1E5)] - expected = BlindspotStatus(0, 0) if sum(raw_speeds) < 1030 else BlindspotStatus(2, 1) - self.assertEqual(pair, create_ioniq6_blindspot_status(0, expected)) - for frame in pair: - self.assertTrue(self.safety.safety_tx_hook(self.packet(frame)), (topology, aol, hex(frame[0]))) + for queued in (False, True): + with self.subTest(topology=topology, aol=aol, speeds=raw_speeds, queued=queued): + _, cp = params(topology) + raw = (RAW_AOL if aol else RAW_LONG)[topology] + cp.safetyConfigs[-1].safetyParam = raw + ci = CarInterface(cp) + controller = CarController(DBC[cp.carFingerprint], cp) + packer = CANPacker(DBC[cp.carFingerprint][Bus.pt]) + bus = CanBus(cp) + inputs = [packer.make_can_msg(name, bus.ECAN, values) for name, values in ( + ("ACCELERATOR", {"GEAR": 5, "COUNTER": 1}), ("TCS", {"COUNTER": 1}), + ("WHEEL_SPEEDS", {"COUNTER": 1}), ("MDPS", {"COUNTER": 1}), + ("CRUISE_BUTTONS", {"COUNTER": 1}), + ("BLINDSPOTS_FRONT_CORNER_2", {"COUNTER": 1, "SIDE_DETECT_STATE": 0x18}), + ("BLINKERS", {"LEFT_LAMP": 1}), + )] + inputs = [frame for frame in inputs if frame[0] != 0xA0] + inputs.append(self.bsm_wheel_frame(packer, raw_speeds, 1)) + inputs.append(packer.make_can_msg("CAM_0x362" if topology == "lkas_alt" else "CAM_0x2a4", bus.CAM, {})) + ci.update([(1_000_000_000, inputs)]) + self.mode(raw) + for frame in inputs: + if frame[2] == bus.ECAN: + self.assertTrue(self.safety.safety_rx_hook(self.packet(frame)), hex(frame[0])) + control = structs.CarControl() + control.leftBlinker = queued + _, sent = controller.update(control.as_reader(), ci.CS, 1_000_000_000) + pair = [frame for frame in sent if frame[0] in (0x1BA, 0x1E5)] + expected = BlindspotStatus(0, 0) if not queued or sum(raw_speeds) < 1030 else BlindspotStatus(2, 1) + self.assertEqual(pair, create_ioniq6_blindspot_status(0, expected)) + for frame in pair: + self.assertTrue(self.safety.safety_tx_hook(self.packet(frame)), (topology, aol, hex(frame[0]))) def test_exhaustive_angle_bit_namespace_and_release_denial(self): for low in range(0x8000):