Compare commits

..

5 Commits

Author SHA1 Message Date
firestar5683 d6934a2da8 Qualify queued Ioniq 6 blind-spot warnings 2026-11-07 17:29:50 -06:00
firestar5683 b9851e59fa Qualify queued Ioniq 6 blind-spot warnings 2026-11-07 17:26:38 -06:00
firestar5683 db987be5c8 Qualify queued Ioniq 6 blind-spot warnings 2026-11-07 17:25:43 -06:00
firestar5683 c5f6a854c5 Qualify queued Ioniq 6 blind-spot warnings 2026-11-07 17:21:46 -06:00
firestar5683 c99cca360d Qualify queued Ioniq 6 blind-spot warnings 2026-11-07 17:19:26 -06:00
43 changed files with 204 additions and 266 deletions
+154
View File
@@ -0,0 +1,154 @@
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)
import re
from collections import Counter
def typing_result():
result = subprocess.run(['.venv/bin/ty', 'check', '--output-format', 'concise', *paths], capture_output=True, text=True)
output = result.stdout + result.stderr
assert result.returncode in (0, 1) and ('error[' in output or result.returncode == 0), output
return result.returncode, output, Counter(re.sub(r':\d+:\d+:', ':', line) for line in output.splitlines() if 'error[' in line)
shipping_typing = typing_result()
saved = {Path(name): Path(name).read_bytes() for name in paths}
try:
for path in saved:
path.write_bytes(subprocess.check_output(['git', 'show', BASE + ':' + str(path)]))
baseline_typing = typing_result()
finally:
for path, data in saved.items():
path.write_bytes(data)
(OUT / 'typing-shipping.txt').write_text(shipping_typing[1])
(OUT / 'typing-baseline.txt').write_text(baseline_typing[1])
assert not (shipping_typing[2] - baseline_typing[2]), shipping_typing[1]
(OUT / 'typing.json').write_text(json.dumps({'introducedErrors': 0,
'baselineErrors': sum(baseline_typing[2].values()), 'shippingErrors': sum(shipping_typing[2].values()),
'baselineExit': baseline_typing[0], 'shippingExit': shipping_typing[0]}))
(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(['readelf', '-sW', 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(['readelf', '-sW', 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('<H', path.read_bytes(), 18)[0] == 183
ctypes.CDLL(str(path.resolve()))
def exports(path):
return {line.split()[-1] for line in subprocess.check_output(['nm', '-D', '--defined-only', str(path)], text=True).splitlines()}
assert exports(original) == exports(target)
deps = subprocess.check_output(['cc', '-MM', '-Ipanda/tests/libpanda', '-Ipanda', '-Ipanda/board', '-Iopendbc_repo',
'panda/tests/libpanda/panda.c'], text=True)
assert 'opendbc/safety/modes/hyundai_canfd.h' in deps
(OUT / 'helper.json').write_text(json.dumps({'sha256': hashlib.sha256(target.read_bytes()).hexdigest(),
'arch': 'AArch64', 'exportsEqual': True, 'dlopen': True, 'includesChangedHeader': True,
'compiler': subprocess.check_output(['cc', '--version'], text=True), 'dependencies': deps}, indent=2))
import tarfile
with tarfile.open(OUT / 'shipping-helper.tar', 'w') as archive:
archive.add(target, arcname=str(target))
if __name__ == '__main__':
{'qualify': qualify, 'firmware': firmware, 'helper': helper}[sys.argv[1]]()
+44
View File
@@ -0,0 +1,44 @@
name: Ioniq 6 queued BSM qualification
on:
push:
branches: [codex/ioniq6-bsm-queued-20261107]
permissions:
contents: read
jobs:
host-native:
runs-on: ubuntu-24.04
timeout-minutes: 35
env:
PYTHONPATH: ${{ github.workspace }}
SCALE: '1'
RAYLIB_BACKEND: headless
steps:
- uses: actions/checkout@v7
- run: git -c gc.auto=0 -c maintenance.auto=false fetch --no-tags --depth=1 --filter=blob:none origin 1d30dc343469e93967d2de3afdc966901009ed5d
- uses: actions/setup-go@v7
with:
go-version: '1.25.1'
cache-dependency-path: mapd_repo/go.sum
- working-directory: mapd_repo
env:
GOTOOLCHAIN: local
run: go mod download
- run: ./tools/op.sh setup
- name: Build matching Linux dependencies
run: >-
.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
+2 -66
View File
@@ -280,65 +280,6 @@ static bool volvo_cma_receiver_quality(const CANPacket_t *msg) {
return volvo_cma_quality(msg) && volvo_cma_lca2_quality(msg);
}
// Unchanged native status is transport evidence, never valid control evidence.
static uint8_t volvo_cma_pscm_status_data[4][8];
static uint32_t volvo_cma_pscm_status_ts[4];
static bool volvo_cma_pscm_status_valid[4];
static uint8_t volvo_cma_pscm_status_index;
static uint8_t volvo_cma_pscm_status_nibble;
static bool volvo_cma_pscm_status_checksum_valid;
static bool volvo_cma_pscm_status_quality_valid;
static void volvo_cma_clear_pscm_status(void) {
for (uint8_t i = 0U; i < 4U; i++) {
volvo_cma_pscm_status_valid[i] = false;
}
volvo_cma_pscm_status_index = 0U;
}
static bool volvo_cma_receiver_quality_rx(const CANPacket_t *msg) {
const bool quality_valid = volvo_cma_receiver_quality(msg);
if (volvo_cma_lca2_mod15 && msg_matches(msg, 0x17U, 2U, 8U) &&
!msg->returned && !msg->rejected && !msg->extended && !msg->fd) {
const uint8_t nibble = msg->data[2] >> 4U;
const bool checksum_valid = volvo_cma_get_checksum(msg) == volvo_cma_compute_checksum(msg);
// Retain the existing four-frame transport tolerance, but never replay an
// earlier status/validation class after a newly observed native transition.
if ((nibble != volvo_cma_pscm_status_nibble) ||
(checksum_valid != volvo_cma_pscm_status_checksum_valid) ||
(quality_valid != volvo_cma_pscm_status_quality_valid)) {
volvo_cma_clear_pscm_status();
}
volvo_cma_pscm_status_nibble = nibble;
volvo_cma_pscm_status_checksum_valid = checksum_valid;
volvo_cma_pscm_status_quality_valid = quality_valid;
const uint8_t slot = volvo_cma_pscm_status_index;
for (uint8_t i = 0U; i < 8U; i++) {
volvo_cma_pscm_status_data[slot][i] = msg->data[i];
}
volvo_cma_pscm_status_ts[slot] = microsecond_timer_get();
volvo_cma_pscm_status_valid[slot] = true;
volvo_cma_pscm_status_index = (slot + 1U) % 4U;
}
return quality_valid;
}
static bool volvo_cma_exact_pscm_status(const CANPacket_t *msg) {
bool matches = false;
if (msg_matches(msg, 0x17U, 0U, 8U) &&
!msg->returned && !msg->rejected && !msg->extended && !msg->fd) {
for (uint8_t slot = 0U; slot < 4U; slot++) {
bool same = volvo_cma_pscm_status_valid[slot] &&
(safety_get_ts_elapsed(microsecond_timer_get(), volvo_cma_pscm_status_ts[slot]) <= 100000U);
for (uint8_t i = 0U; i < 8U; i++) {
same &= msg->data[i] == volvo_cma_pscm_status_data[slot][i];
}
matches |= same;
}
}
return matches;
}
// The Polestar source clocks advance nominally2/4 per50Hz frame, with
// observed sampling jitter of1..3/3..5. Frozen/jumped counters retain the shared tolerance.
static uint8_t volvo_cma_lca2_counter(const CANPacket_t *msg) {
@@ -442,8 +383,7 @@ static bool volvo_cma_tx_hook(const CANPacket_t *msg) {
// policy or intent and retains the existing health deadlines/tolerances.
bool tx;
if (!controls_allowed || !volvo_cma_drive_current() || !aol_rx_healthy()) {
tx = (volvo_cma_lca2_mod15 && (msg->addr == 0x17U)) ?
volvo_cma_exact_pscm_status(msg) : volvo_cma_exact_stock(msg);
tx = volvo_cma_exact_stock(msg);
} else {
tx = volvo_cma_receiver_quality(msg) && volvo_cma_source_projection(msg);
if ((msg->addr == 0x69U) || (msg->addr == 0x67U) || (msg->addr == 0x17U)) {
@@ -574,10 +514,6 @@ static RxCheck *volvo_cma_rx_checks_workspace(void) {
static safety_config volvo_cma_init(uint16_t param) {
RxCheck *const rx_workspace = volvo_cma_rx_checks_workspace();
volvo_cma_lca2_mod15 = param == 5U;
volvo_cma_clear_pscm_status();
volvo_cma_pscm_status_nibble = 0U;
volvo_cma_pscm_status_checksum_valid = false;
volvo_cma_pscm_status_quality_valid = false;
volvo_cma_lca2_second_seen = false;
volvo_cma_lca2_second_last = 0U;
volvo_cma_lca2_second_wrong = 0;
@@ -699,6 +635,6 @@ const safety_hooks volvo_cma_hooks = {
.get_counter = volvo_cma_lca2_counter,
.get_checksum = volvo_cma_get_checksum,
.compute_checksum = volvo_cma_compute_checksum,
.get_quality_flag_valid = volvo_cma_receiver_quality_rx,
.get_quality_flag_valid = volvo_cma_receiver_quality,
// No custom fwd hook - stock LCA always blocked by .check_relay = true
};
@@ -29,7 +29,7 @@ static uint32_t volvo_dispatch_compute_checksum(const CANPacket_t *msg) {
return volvo_dispatch_c1 ? 0U : volvo_cma_compute_checksum(msg);
}
static bool volvo_dispatch_quality(const CANPacket_t *msg) {
return volvo_dispatch_c1 || volvo_cma_receiver_quality_rx(msg);
return volvo_dispatch_c1 || volvo_cma_receiver_quality(msg);
}
const safety_hooks volvo_dispatch_hooks = {.init = volvo_dispatch_init, .rx = volvo_dispatch_rx, .tx = volvo_dispatch_tx,
.get_counter = volvo_dispatch_get_counter,
@@ -86,199 +86,3 @@ class TestPolestarCounters(unittest.TestCase):
self.assertTrue(self.safety.safety_config_valid())
self.assertTrue(self.safety.get_controls_allowed())
self.assertTrue(self.tx(self.lca_command()))
def unknown_pscm_status(self):
return 0x17, bytes.fromhex('d400a0858000ffff'), 2
def test_unknown_status_relays_without_becoming_control_evidence(self):
for cruise in (False, True):
with self.subTest(cruise=cruise):
self.reset()
incoming = self.feed(cruise=cruise)
status = self.unknown_pscm_status()
self.assertFalse(self.rx(status))
self.assertFalse(self.safety.get_controls_allowed())
self.safety.safety_tick()
self.assertFalse(self.safety.safety_config_valid())
self.assertTrue(self.tx(self.relay(status)))
self.assertFalse(self.tx(self.relay(incoming[0x17])))
self.assertFalse(self.tx(self.lca_command()))
self.safety.set_controls_allowed(True)
self.assertTrue(self.tx(self.relay(status)))
self.assertFalse(self.tx(self.lca_command()))
for address in cma.RELAY_BUSES:
if address != 0x17:
self.assertTrue(self.tx(self.relay(incoming[address])))
self.assertEqual(self.safety.safety_fwd_hook(2, 0x17), -1)
def test_unknown_status_requires_exact_native_bytes(self):
self.feed()
status = self.unknown_pscm_status()
self.assertFalse(self.rx(status))
for bit in range(64):
data = bytearray(status[1])
data[bit // 8] ^= 1 << (bit % 8)
self.assertFalse(self.tx((0x17, data, 0)), bit)
self.assertTrue(self.tx(self.relay(status)))
def test_unknown_status_source_identity_cannot_be_forged(self):
status = self.unknown_pscm_status()
for field in ('returned', 'rejected', 'extended', 'fd'):
with self.subTest(field=field):
self.reset()
self.feed(cruise=False)
packet = cma.libsafety_py.make_CANPacket(status[0], 2, status[1])
setattr(packet[0], field, 1)
self.safety.safety_rx_hook(packet)
self.assertFalse(self.tx(self.relay(status)))
for bus in (0, 1, 3, 4, 5, 6, 7):
self.reset()
self.feed(cruise=False)
self.rx((0x17, status[1], bus))
self.assertFalse(self.tx(self.relay(status)))
for length in (*range(8), 12):
self.reset()
self.feed(cruise=False)
self.rx((0x17, (status[1] + bytes(4))[:length], 2))
self.assertFalse(self.tx(self.relay(status)))
def test_unknown_status_tx_cannot_seed_or_refresh_source(self):
self.feed(cruise=False)
status = self.unknown_pscm_status()
self.assertFalse(self.tx(self.relay(status)))
self.assertFalse(self.rx(status))
self.safety.set_timer(110000)
self.assertTrue(self.tx(self.relay(status)))
self.safety.set_timer(110001)
self.assertFalse(self.tx(self.relay(status)))
self.assertFalse(self.tx(self.relay(status)))
def test_unknown_status_original_age_survives_timer_wrap(self):
self.feed(cruise=False)
start = 0xFFFFFF00
self.safety.set_timer(start)
status = self.unknown_pscm_status()
self.assertFalse(self.rx(status))
for age, accepted in ((90000, True), (100000, True), (100001, False)):
self.safety.set_timer((start + age) & 0xFFFFFFFF)
self.assertEqual(bool(self.tx(self.relay(status))), accepted)
def test_status_preserves_four_frame_transport_tolerance(self):
for unknown in (False, True):
with self.subTest(unknown=unknown):
self.reset()
incoming = self.feed(cruise=False)
source = self.unknown_pscm_status() if unknown else incoming[0x17]
frames = [(0x17, source[1][:-1] + bytes([i]), 2) for i in range(5)]
for i, frame in enumerate(frames):
self.safety.set_timer(20000 + i * 10000)
self.assertEqual(bool(self.rx(frame)), not unknown)
for retained in frames[max(0, i - 3):i + 1]:
self.assertTrue(self.tx(self.relay(retained)))
self.assertFalse(self.tx(self.relay(frames[0])))
# A new native arrival never refreshes another slot's original age.
self.safety.set_timer(130000)
self.assertTrue(self.tx(self.relay(frames[1])))
self.safety.set_timer(130001)
self.assertFalse(self.tx(self.relay(frames[1])))
self.assertTrue(self.tx(self.relay(frames[2])))
def test_native_status_transition_flushes_old_transport_evidence(self):
incoming = self.feed(cruise=False)
healthy = incoming[0x17]
first = self.unknown_pscm_status()
second = first[0], first[1][:-1] + b'\xfe', first[2]
restored = healthy[0], healthy[1][:-1] + b'\x01', healthy[2]
self.assertFalse(self.rx(first))
self.assertTrue(self.tx(self.relay(first)))
self.assertFalse(self.tx(self.relay(healthy)))
self.assertFalse(self.rx(second))
self.assertTrue(self.tx(self.relay(first)))
self.assertTrue(self.tx(self.relay(second)))
self.assertTrue(self.rx(restored))
for stale in (healthy, first, second):
self.assertFalse(self.tx(self.relay(stale)))
self.assertTrue(self.tx(self.relay(restored)))
self.assertFalse(self.safety.get_controls_allowed())
def test_each_native_validation_class_change_flushes_transport_evidence(self):
healthy = self.feed(cruise=False)[0x17]
# Keep the high nibble at 8 and change one validation boolean at a time:
# (checksum, quality): TT -> FT -> FF -> TF -> TT.
bad_checksum = bytearray(healthy[1])
bad_checksum[0] ^= 1
bad_both = bytearray(bad_checksum)
bad_both[1] |= 8 # This quality bit is unconstrained by the checksum.
bad_quality = bytearray(healthy[1])
bad_quality[1] |= 8
frames = [healthy, (0x17, bad_checksum, 2), (0x17, bad_both, 2),
(0x17, bad_quality, 2), (0x17, healthy[1][:-1] + b'\x01', 2)]
for i, frame in enumerate(frames):
self.assertEqual(bool(self.rx(frame)), i in (0, 4))
self.assertTrue(self.tx(self.relay(frame)))
for stale in frames[:i]:
self.assertFalse(self.tx(self.relay(stale)))
self.assertFalse(self.safety.get_controls_allowed())
def test_unknown_status_cache_does_not_cross_mode_reset(self):
self.feed(cruise=False)
status = self.unknown_pscm_status()
self.assertFalse(self.rx(status))
self.assertTrue(self.tx(self.relay(status)))
self.reset()
self.assertFalse(self.tx(self.relay(status)))
self.feed(cruise=False)
self.assertFalse(self.tx(self.relay(status)))
def test_unknown_status_does_not_expand_other_profiles(self):
status = self.unknown_pscm_status()
for profile in (0, 1):
self.assertEqual(self.safety.set_safety_hooks(36, profile), 0)
self.safety.init_tests()
self.safety.set_timer(10000)
self.assertFalse(self.rx(status))
self.assertFalse(self.tx(self.relay(status)))
self.assertNotEqual(self.safety.set_safety_hooks(36, 4), 0)
def test_unknown_status_remains_blocked_on_relay_malfunction(self):
self.feed(cruise=False)
status = self.unknown_pscm_status()
self.assertFalse(self.rx(status))
self.assertTrue(self.tx(self.relay(status)))
self.rx((0x58, bytes(8), 2))
self.assertTrue(self.safety.get_relay_malfunction())
self.assertFalse(self.tx(self.relay(status)))
def test_unknown_status_tx_identity_matches_native_framing(self):
self.feed(cruise=False)
status = self.unknown_pscm_status()
self.assertFalse(self.rx(status))
for field in ('returned', 'rejected', 'extended', 'fd'):
packet = cma.libsafety_py.make_CANPacket(0x17, 0, status[1])
setattr(packet[0], field, 1)
self.assertFalse(self.safety.safety_tx_hook(packet), field)
self.assertTrue(self.tx(self.relay(status)))
for bus in range(1, 8):
self.assertFalse(self.tx((0x17, status[1], bus)))
for length in (*range(8), 12):
self.assertFalse(self.tx((0x17, (status[1] + bytes(4))[:length], 0)))
self.assertFalse(self.tx((0x16, status[1], 0)))
self.assertTrue(self.tx(self.relay(status)))
def test_modified_status_is_rejected_until_controller_returns_to_stock(self):
self.feed()
status = self.unknown_pscm_status()
stock = {
'CHECKSUM_1': 0xD4, 'LCA_ENABLED_ECHO': 0, 'SIG1_BYTE_1_HI_NIBBLE': 0,
'SIG1_REPLICA_BYTE_2_LO_NIBLE': 0, 'NEW_SIGNAL_2': 10, 'CHECKSUM_2': 0x85,
'BYTE_4_5': 0x8000, 'BYTE_6': 0xFF, 'BYTE_7': 0xFF,
}
self.assertFalse(self.rx(status))
self.assertFalse(self.safety.get_controls_allowed())
modified = cma.cma_can.create_pscm_related_message(self.mid, True, False, stock, 1)
self.assertNotEqual(modified[1], status[1])
self.assertFalse(self.tx(modified))
unchanged = cma.cma_can.create_pscm_related_message(self.mid, False, False, stock, 1)
self.assertEqual(unchanged[1], status[1])
self.assertTrue(self.tx(unchanged))
self.assertFalse(self.safety.get_controls_allowed())
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+1 -1
View File
@@ -1,4 +1,4 @@
#pragma once
extern const uint8_t gitversion[29];
const uint8_t gitversion[29] = "DEV-3093e5b7-Pf38f504f-DEBUG";
const uint8_t gitversion[29] = "DEV-5d6e6afc-Pf38f504f-DEBUG";
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+1 -1
View File
@@ -1 +1 @@
DEV-3093e5b7-Pf38f504f-DEBUG
DEV-5d6e6afc-Pf38f504f-DEBUG
Binary file not shown.
+1 -1
View File
@@ -1 +1 @@
{"protected_inventory_sha256": "f1741daeb7b18776c77b38c839c9a6e97bfb97229851365246afdf6e28936f60", "schema_version": 1}
{"protected_inventory_sha256": "8aa2700f218ddb93efedb1749b7d7edc2a0992e858f4ae695124078b7f8c1117", "schema_version": 1}