mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-28 17:13:54 +08:00
Restore Volvo safety coverage and correct new test fixtures
This commit is contained in:
Dom
@@ -7,6 +7,9 @@ from opendbc.car.fingerprints import FW_VERSIONS
|
||||
from opendbc.car.fw_versions import FW_QUERY_CONFIGS
|
||||
from opendbc.car.interfaces import CarInterfaceBase, get_interface_attr
|
||||
from opendbc.car.values import PLATFORMS
|
||||
from opendbc.car.volvo.cma_carcontroller import CarControllerCMA
|
||||
from opendbc.car.volvo.cma_carstate import CarStateCMA
|
||||
from opendbc.car.volvo.values import CAR as VOLVO
|
||||
from opendbc.testing import Fuzzy, fuzzy_test
|
||||
|
||||
ALL_ECUS = tuple(sorted({ecu for ecus in FW_VERSIONS.values() for ecu in ecus} |
|
||||
@@ -112,8 +115,12 @@ def _make_optional_hardware_test(car_name):
|
||||
self.assertEqual(params.carFingerprint, car_name)
|
||||
self.assertEqual(fingerprint, gen_empty_fingerprint())
|
||||
interface = interface_type(params)
|
||||
self.assertIsInstance(interface.CS, interface_type.CarState)
|
||||
self.assertIsInstance(interface.CC, interface_type.CarController)
|
||||
if car_name in (VOLVO.VOLVO_XC40_RECHARGE, VOLVO.VOLVO_S60_RECHARGE, VOLVO.POLESTAR_2):
|
||||
self.assertIsInstance(interface.CS, CarStateCMA)
|
||||
self.assertIsInstance(interface.CC, CarControllerCMA)
|
||||
else:
|
||||
self.assertIsInstance(interface.CS, interface_type.CarState)
|
||||
self.assertIsInstance(interface.CC, interface_type.CarController)
|
||||
|
||||
return test
|
||||
|
||||
|
||||
@@ -0,0 +1,373 @@
|
||||
"""Native CMA/SPA contracts, also collected by the safety unittest coverage gate.
|
||||
|
||||
The fixtures use the same DBCs and checksum encoders as the existing Volvo car
|
||||
integration tests. Permissions come from physical RX, never a forced active flag.
|
||||
"""
|
||||
import unittest
|
||||
|
||||
from opendbc.can import CANPacker
|
||||
from opendbc.car.volvo import cma_can
|
||||
from opendbc.car.volvo.helpers import (
|
||||
checksum_lca_2_message, checksum_2_0x69_message, checksum_lca_5_message,
|
||||
checksum_1_pscm_related_message, checksum_2_pscm_related_message,
|
||||
)
|
||||
from opendbc.safety.tests.libsafety import libsafety_py
|
||||
|
||||
|
||||
RELAY_BUSES = {0x58: 2, 0x57: 2, 0x69: 2, 0x90: 2, 0x67: 2, 0x97: 2, 0x92: 2, 0x16: 0, 0x17: 0}
|
||||
|
||||
|
||||
def checksummed(message):
|
||||
address, payload, bus = message
|
||||
data = bytearray(payload)
|
||||
if address == 0x69:
|
||||
data[6] = checksum_lca_2_message(data[0], data[5])
|
||||
data[2] = checksum_2_0x69_message(data[0], data[1], data[3], data[4])
|
||||
elif address == 0x67:
|
||||
data[2] = checksum_lca_5_message(data[0], data[1], data[3], data[4], data[5])
|
||||
elif address == 0x17:
|
||||
data[0] = checksum_1_pscm_related_message(data[1], data[2])
|
||||
data[3] = checksum_2_pscm_related_message(data[2])
|
||||
return address, bytes(data), bus
|
||||
|
||||
|
||||
class TestVolvoCMASafety(unittest.TestCase):
|
||||
PARAM = 0
|
||||
|
||||
def setUp(self):
|
||||
self.safety = libsafety_py.libsafety
|
||||
self.mid = CANPacker('volvo_mid_1')
|
||||
self.pt = CANPacker('volvo_front_1_spa' if self.PARAM else 'volvo_front_1_cma')
|
||||
self.reset()
|
||||
|
||||
def reset(self):
|
||||
self.assertEqual(self.safety.set_safety_hooks(36, self.PARAM), 0)
|
||||
self.safety.init_tests()
|
||||
self.safety.set_timer(10000)
|
||||
|
||||
def rx(self, message):
|
||||
address, data, bus = message
|
||||
return self.safety.safety_rx_hook(libsafety_py.make_CANPacket(address, bus, data))
|
||||
|
||||
def tx(self, message):
|
||||
address, data, bus = message
|
||||
return self.safety.safety_tx_hook(libsafety_py.make_CANPacket(address, bus, data))
|
||||
|
||||
def sources(self, cruise=True, gear=3, sample=0, parity=0):
|
||||
definitions: list[tuple[str, int, dict[str, float]]] = [
|
||||
('SPEED', 0, {'SPEED': 3000}), ('SPEED_2', 0, {}), ('GEAR_POSITION', 0, {'GEAR_POSITION': gear}),
|
||||
('LCA', 0, {}), ('LCA_2', 0, {'COUNTER_1': (sample * 2 + parity) % 16, 'COUNTER_2': sample * 4 % 16}),
|
||||
('LCA_3', 0, {}), ('LCA_4', 0, {}), ('LCA_5', 0, {}), ('LCA_6', 0, {}), ('LCA_7', 0, {}),
|
||||
('PSCM', 2, {}), ('DRIVER_INPUT', 2, {}), ('EGSM', 2, {}),
|
||||
('PSCM_RELATED', 2, {'NEW_SIGNAL_2': 8}), ('SAS', 2, {}),
|
||||
]
|
||||
messages = [self.mid.make_can_msg(name, bus, values) for name, bus, values in definitions]
|
||||
messages.append(self.pt.make_can_msg('ECM_1', 1, {'ACCELERATOR_PEDAL_POS': 0 if self.PARAM else 20}))
|
||||
cruise_signal = 'CRUISE_CONTROL_SPA_ENABLED' if self.PARAM else 'CRUISE_CONTROL_ENABLED'
|
||||
messages.append(self.pt.make_can_msg('BUS1_CRUISE_CONTROL', 1, {cruise_signal: int(cruise)}))
|
||||
return [checksummed(message) for message in messages]
|
||||
|
||||
def feed(self, **kwargs):
|
||||
incoming = self.sources(**kwargs)
|
||||
for message in incoming:
|
||||
self.assertTrue(self.rx(message), message)
|
||||
self.safety.safety_tick()
|
||||
self.assertTrue(self.safety.safety_config_valid())
|
||||
self.assertEqual(bool(self.safety.get_controls_allowed()), kwargs.get('cruise', True))
|
||||
return {address: (address, payload, bus) for address, payload, bus in incoming}
|
||||
|
||||
def relay(self, source):
|
||||
address, data, _ = source
|
||||
return address, data, RELAY_BUSES[address]
|
||||
|
||||
def lca_command(self, positive=614, negative=-614):
|
||||
return cma_can.create_lca_message(self.mid, True, 0, {
|
||||
'LCA_STEER': 0, 'OPAQUE_BYTE_0': 0, 'OPAQUE_BYTE_2': 0, 'OPAQUE_BYTE_7': 0,
|
||||
}, authority_pos=positive, authority_neg=negative)
|
||||
|
||||
def angle_command(self, raw):
|
||||
# LCA_5_STEER is signed 15-bit, 0.05596 degrees/count, in bytes 6-7.
|
||||
data = bytearray(8)
|
||||
data[6:8] = (raw & 0x7FFF).to_bytes(2, 'big')
|
||||
return checksummed((0x67, data, 2))
|
||||
|
||||
def test_rx_sensor_scales_and_profile_pedals(self):
|
||||
self.feed()
|
||||
self.assertAlmostEqual(self.safety.get_vehicle_speed_max(), 3000 * .003977, places=5)
|
||||
for raw in (-100, 100):
|
||||
for _ in range(6):
|
||||
self.assertTrue(self.rx(self.mid.make_can_msg('PSCM', 2, {'PSCM_ANGLE_SENSOR': raw * .05596})))
|
||||
self.assertEqual(self.safety.get_angle_meas_min(), raw)
|
||||
self.assertEqual(self.safety.get_angle_meas_max(), raw)
|
||||
for torque in (-10, 10):
|
||||
for _ in range(6):
|
||||
self.assertTrue(self.rx(self.mid.make_can_msg('DRIVER_INPUT', 2, {'STEERING_DRIVER_INPUT': torque})))
|
||||
self.assertEqual(self.safety.get_torque_driver_min(), torque)
|
||||
self.assertEqual(self.safety.get_torque_driver_max(), torque)
|
||||
threshold = 1 if self.PARAM else 21
|
||||
step = 1 / 256 if self.PARAM else 1
|
||||
for value, pressed in ((threshold + step, True), (threshold, False)):
|
||||
self.assertTrue(self.rx(self.pt.make_can_msg('ECM_1', 1, {'ACCELERATOR_PEDAL_POS': value})))
|
||||
self.assertEqual(bool(self.safety.get_gas_pressed_prev()), pressed)
|
||||
for sample, pressed in enumerate((True, False), 1):
|
||||
message = self.mid.make_can_msg('LCA_2', 0, {'COUNTER_1': sample * 2, 'COUNTER_2': sample * 4,
|
||||
'BRAKE_PEDAL_PRESSED_B': int(pressed)})
|
||||
self.assertTrue(self.rx(checksummed(message)))
|
||||
self.assertEqual(bool(self.safety.get_brake_pressed_prev()), pressed)
|
||||
self.assertFalse(self.safety.get_controls_allowed())
|
||||
|
||||
def test_stock_cruise_withdrawal_and_reengagement(self):
|
||||
for enabled in (False, True, False, True):
|
||||
signal = 'CRUISE_CONTROL_SPA_ENABLED' if self.PARAM else 'CRUISE_CONTROL_ENABLED'
|
||||
self.assertTrue(self.rx(self.pt.make_can_msg('BUS1_CRUISE_CONTROL', 1, {signal: int(enabled)})))
|
||||
self.assertEqual(bool(self.safety.get_controls_allowed()), enabled)
|
||||
if not self.PARAM:
|
||||
self.assertTrue(self.rx(self.pt.make_can_msg('BUS1_CRUISE_CONTROL', 1, {})))
|
||||
self.assertTrue(self.rx(self.pt.make_can_msg('BUS1_CRUISE_CONTROL', 1, {'CRUISE_CONTROL_ENABLED_IDLE_TRAFFIC': 1})))
|
||||
self.assertTrue(self.safety.get_controls_allowed())
|
||||
|
||||
def test_stock_only_when_cruise_withdrawn_and_original_age_expires(self):
|
||||
incoming = self.feed(cruise=False)
|
||||
for address in RELAY_BUSES:
|
||||
message = self.relay(incoming[address])
|
||||
self.assertTrue(self.tx(message))
|
||||
data = bytearray(message[1])
|
||||
data[0] ^= 1
|
||||
self.assertFalse(self.tx((address, data, message[2])))
|
||||
self.safety.set_timer(110000)
|
||||
for address in RELAY_BUSES:
|
||||
self.assertTrue(self.tx(self.relay(incoming[address])))
|
||||
self.safety.set_timer(110001)
|
||||
for address in RELAY_BUSES:
|
||||
self.assertFalse(self.tx(self.relay(incoming[address])))
|
||||
|
||||
def test_allowlist_pins_bus_and_length(self):
|
||||
incoming = self.feed(cruise=False)
|
||||
for address, expected_bus in RELAY_BUSES.items():
|
||||
for bus in range(8):
|
||||
self.assertEqual(bool(self.tx((address, incoming[address][1], bus))), bus == expected_bus)
|
||||
for length in range(8):
|
||||
self.assertFalse(self.tx((address, incoming[address][1][:length], bus)))
|
||||
for address in (0x15, 0x55, 0x60, 0x68, 0x80, 0x45):
|
||||
for bus in range(3):
|
||||
self.assertFalse(self.tx((address, bytes(8), bus)))
|
||||
|
||||
def test_active_native_source_bound_commands(self):
|
||||
incoming = self.feed()
|
||||
for address in RELAY_BUSES:
|
||||
self.assertTrue(self.tx(self.relay(incoming[address])))
|
||||
self.assertTrue(self.tx(self.lca_command()))
|
||||
self.assertTrue(self.tx(self.angle_command(1)))
|
||||
# These two hands-on relay payloads are emitted by create_pscm_message.
|
||||
for hands_a, hands_b in ((195, 186), (249, 154)):
|
||||
message = self.mid.make_can_msg('PSCM', 0, {'HANDS_ON_STEERING_WHEEL_A': hands_a,
|
||||
'HANDS_ON_STEERING_WHEEL_B': hands_b})
|
||||
self.assertTrue(self.tx(message))
|
||||
# Counter replication may advance, but its observed echo must be preserved or zeroed.
|
||||
for counter in range(15):
|
||||
message = self.mid.make_can_msg('PSCM_RELATED', 0, {'NEW_SIGNAL_2': 8,
|
||||
'SIG1_BYTE_1_HI_NIBBLE': counter, 'SIG1_REPLICA_BYTE_2_LO_NIBLE': counter})
|
||||
self.assertTrue(self.tx(checksummed(message)))
|
||||
|
||||
def test_opaque_stock_fields_cannot_be_forged_with_recomputed_checksums(self):
|
||||
incoming = self.feed()
|
||||
# Start from accepted active templates, so an opaque mutation cannot be
|
||||
# rejected merely because the stock frame lacks the fixed active fields.
|
||||
active = {
|
||||
0x58: self.lca_command(),
|
||||
0x57: cma_can.create_lca_3_message(self.mid, True, 0, {'SPEED_A': 0, 'SPEED_B': 0, 'NEW_SIGNAL_9': 0}, 0),
|
||||
0x69: checksummed(self.mid.make_can_msg('LCA_2', 2, {'BYTE_0': 24, 'PILOT_ASSIST_ENGAGED': 1, 'NEW_SIGNAL_3': 3})),
|
||||
0x90: self.mid.make_can_msg('LCA_4', 2, {'LCA_ENABLE': 3}),
|
||||
0x67: self.angle_command(1),
|
||||
0x16: self.mid.make_can_msg('PSCM', 0, {'HANDS_ON_STEERING_WHEEL_A': 195, 'HANDS_ON_STEERING_WHEEL_B': 186}),
|
||||
0x17: checksummed(self.mid.make_can_msg('PSCM_RELATED', 0, {'NEW_SIGNAL_2': 8,
|
||||
'SIG1_BYTE_1_HI_NIBBLE': 1, 'SIG1_REPLICA_BYTE_2_LO_NIBLE': 1})),
|
||||
}
|
||||
for address, byte, mask in ((0x58, 6, 1), (0x57, 2, 1), (0x69, 5, 0x40), (0x90, 4, 1),
|
||||
(0x67, 0, 1), (0x97, 0, 1), (0x92, 0, 1), (0x16, 7, 1), (0x17, 6, 1)):
|
||||
with self.subTest(address=address):
|
||||
baseline = active.get(address, self.relay(incoming[address]))
|
||||
self.assertTrue(self.tx(baseline))
|
||||
data = bytearray(baseline[1])
|
||||
data[byte] ^= mask
|
||||
self.assertFalse(self.tx(checksummed((address, data, RELAY_BUSES[address]))))
|
||||
|
||||
def test_torque_authority_limits(self):
|
||||
self.feed()
|
||||
for positive, negative, allowed in ((0, 0, True), (614, -614, True), (615, -614, False),
|
||||
(614, -615, False), (-1, 0, False), (0, 1, False)):
|
||||
self.assertEqual(bool(self.tx(self.lca_command(positive, negative))), allowed)
|
||||
|
||||
def test_angle_absolute_and_rate_limits(self):
|
||||
for raw in (-9651, 9651):
|
||||
self.reset()
|
||||
self.feed()
|
||||
self.safety.set_desired_angle_last(raw)
|
||||
self.assertFalse(self.tx(self.angle_command(raw)))
|
||||
for raw in (-1, 1):
|
||||
self.reset()
|
||||
self.feed()
|
||||
self.assertTrue(self.tx(self.angle_command(raw)))
|
||||
self.reset()
|
||||
self.feed()
|
||||
self.assertFalse(self.tx(self.angle_command(1000)))
|
||||
|
||||
def test_pscm_relay_cannot_invent_measured_angle(self):
|
||||
self.feed()
|
||||
for raw in (-55, -54, 54, 55):
|
||||
self.assertTrue(self.rx(self.mid.make_can_msg('PSCM', 2, {'PSCM_ANGLE_SENSOR': raw * .05596})))
|
||||
# Keep the source-bound angle intact and isolate the measurement guard.
|
||||
# The standard measurement fixture does not grant control permission.
|
||||
self.safety.set_angle_meas(0, 0)
|
||||
message = self.mid.make_can_msg('PSCM', 0, {'PSCM_ANGLE_SENSOR': raw * .05596,
|
||||
'HANDS_ON_STEERING_WHEEL_A': 195, 'HANDS_ON_STEERING_WHEEL_B': 186})
|
||||
self.assertEqual(bool(self.tx(message)), abs(raw) <= 54)
|
||||
for hands_a, hands_b in ((195, 154), (249, 186), (0, 1)):
|
||||
self.assertFalse(self.tx(self.mid.make_can_msg('PSCM', 0, {
|
||||
'HANDS_ON_STEERING_WHEEL_A': hands_a, 'HANDS_ON_STEERING_WHEEL_B': hands_b})))
|
||||
|
||||
def test_gear_is_physical_current_and_drive_only(self):
|
||||
for gear in range(8):
|
||||
with self.subTest(gear=gear):
|
||||
self.reset()
|
||||
incoming = self.feed(gear=gear)
|
||||
self.assertEqual(bool(self.tx(self.lca_command())), gear in (3, 4))
|
||||
self.assertTrue(self.tx(self.relay(incoming[0x58])))
|
||||
self.reset()
|
||||
self.feed()
|
||||
self.safety.set_timer(110001)
|
||||
for message in self.sources(sample=1):
|
||||
if message[0] != 0x80:
|
||||
self.assertTrue(self.rx(message))
|
||||
self.assertTrue(self.safety.get_controls_allowed())
|
||||
self.assertFalse(self.tx(self.lca_command()))
|
||||
|
||||
def test_each_known_checksum_rejects_corrupt_rx_and_tx(self):
|
||||
for address, index in ((0x69, 6), (0x69, 2), (0x67, 2), (0x17, 0), (0x17, 3)):
|
||||
with self.subTest(address=address, index=index):
|
||||
self.reset()
|
||||
incoming = self.feed()
|
||||
data = bytearray(incoming[address][1])
|
||||
data[index] ^= 1
|
||||
self.assertFalse(self.tx((address, data, RELAY_BUSES[address])))
|
||||
self.assertFalse(self.rx((address, data, incoming[address][2])))
|
||||
self.assertFalse(self.safety.get_controls_allowed())
|
||||
self.assertFalse(self.tx(self.lca_command()))
|
||||
self.assertTrue(self.tx(self.relay(incoming[0x58])))
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_config_valid())
|
||||
|
||||
def test_unknown_pscm_counter_domains_rejected_despite_correct_checksum(self):
|
||||
for first, second in ((0x10, 0x80), (0xF0, 0x8F), (8, 0x80), (0, 0)):
|
||||
self.reset()
|
||||
self.feed()
|
||||
data = bytearray(8)
|
||||
data[1:3] = bytes((first, second))
|
||||
message = checksummed((0x17, data, 0))
|
||||
self.assertFalse(self.tx(message))
|
||||
self.assertFalse(self.rx((message[0], message[1], 2)))
|
||||
|
||||
def test_documented_lca2_counter_wraps_but_frozen_counter_expires(self):
|
||||
for parity in (0, 1):
|
||||
self.reset()
|
||||
self.feed(parity=parity)
|
||||
for sample in range(1, 25):
|
||||
self.safety.set_timer(10000 + sample * 20000)
|
||||
self.feed(sample=sample, parity=parity)
|
||||
frozen = next(message for message in self.sources(sample=24, parity=parity) if message[0] == 0x69)
|
||||
for sample in range(25, 35):
|
||||
self.safety.set_timer(10000 + sample * 20000)
|
||||
for message in self.sources(sample=sample, parity=parity):
|
||||
if message[0] == 0x69:
|
||||
self.rx(frozen) # The shared wrong-counter tolerance is eventually exhausted.
|
||||
else:
|
||||
self.assertTrue(self.rx(message))
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_config_valid())
|
||||
self.assertFalse(self.safety.get_controls_allowed())
|
||||
|
||||
def test_lca2_phase_must_come_from_checksum_valid_physical_rx(self):
|
||||
for index, mask in ((1, 1), (5, 1), (5, 4)):
|
||||
self.reset()
|
||||
incoming = self.feed()
|
||||
data = bytearray(incoming[0x69][1])
|
||||
data[index] ^= mask
|
||||
self.assertFalse(self.rx(checksummed((0x69, data, 0))))
|
||||
self.assertFalse(self.safety.get_controls_allowed())
|
||||
self.reset()
|
||||
message = next(message for message in self.sources() if message[0] == 0x69)
|
||||
data = bytearray(message[1])
|
||||
data[6] ^= 1
|
||||
self.assertFalse(self.rx((0x69, data, 0)))
|
||||
self.assertFalse(self.tx(self.relay(message)))
|
||||
self.feed(parity=1)
|
||||
|
||||
def test_existing_lca2_special_case_checksum(self):
|
||||
message = self.mid.make_can_msg('LCA_2', 0, {'BRAKE_PEDAL_PRESSED_A': 0})
|
||||
self.assertEqual(message[1][0], 0)
|
||||
self.assertEqual(message[1][5], 128)
|
||||
self.assertEqual(checksummed(message)[1][6], 0)
|
||||
self.assertTrue(self.rx(checksummed(message)))
|
||||
|
||||
def test_every_required_receiver_needed_after_mode_reset(self):
|
||||
for missing in range(17):
|
||||
self.reset()
|
||||
incoming = self.feed()
|
||||
self.assertTrue(self.tx(self.lca_command()))
|
||||
self.reset()
|
||||
self.assertFalse(self.tx(self.relay(incoming[0x58])))
|
||||
messages = self.sources()
|
||||
for index, message in enumerate(messages):
|
||||
if index != missing:
|
||||
self.assertTrue(self.rx(message))
|
||||
self.assertFalse(self.safety.safety_config_valid())
|
||||
self.assertFalse(self.tx(self.lca_command()))
|
||||
self.assertTrue(self.rx(messages[missing]))
|
||||
self.assertTrue(self.safety.safety_config_valid())
|
||||
self.assertTrue(self.tx(self.lca_command()))
|
||||
|
||||
def test_required_receiver_loss_rejects_modified_commands(self):
|
||||
for omitted in range(17):
|
||||
self.reset()
|
||||
self.feed()
|
||||
for sample in range(1, 57):
|
||||
self.safety.set_timer(10000 + sample * 20000)
|
||||
for index, message in enumerate(self.sources(sample=sample)):
|
||||
if index != omitted:
|
||||
self.assertTrue(self.rx(message))
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_config_valid())
|
||||
self.assertFalse(self.safety.get_controls_allowed())
|
||||
self.assertFalse(self.tx(self.lca_command()))
|
||||
|
||||
def test_stock_relay_history_does_not_cross_mode_reset(self):
|
||||
incoming = self.feed(cruise=False)
|
||||
first = incoming[0x97]
|
||||
for value in range(1, 5):
|
||||
message = (0x97, bytes([value]) + bytes(7), 0)
|
||||
self.assertTrue(self.rx(message))
|
||||
self.assertTrue(self.tx(self.relay(message)))
|
||||
self.assertFalse(self.tx(self.relay(first)))
|
||||
self.reset()
|
||||
self.assertFalse(self.tx(self.relay(message)))
|
||||
|
||||
def test_relay_malfunction_and_forwarding(self):
|
||||
self.feed()
|
||||
for bus in (0, 2):
|
||||
for address in (*RELAY_BUSES, 0x60):
|
||||
blocked = RELAY_BUSES.get(address) == 2 - bus
|
||||
self.assertEqual(self.safety.safety_fwd_hook(bus, address), -1 if blocked else 2 - bus)
|
||||
for address, bus in RELAY_BUSES.items():
|
||||
self.reset()
|
||||
self.assertTrue(self.rx((address, bytes(8), bus)))
|
||||
self.assertTrue(self.safety.get_relay_malfunction())
|
||||
self.assertFalse(self.tx(self.lca_command()))
|
||||
self.assertEqual(self.safety.safety_fwd_hook(0, 0x60), -1)
|
||||
|
||||
|
||||
class TestVolvoSPASafety(TestVolvoCMASafety):
|
||||
PARAM = 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -49,9 +49,9 @@ def test_recorded_contract_is_exact_original_comparison_for_all_signs():
|
||||
def test_bad_support_versions_and_unversioned_optin_fail_closed():
|
||||
for position in range(3):
|
||||
for bad in (math.nan, math.inf, True):
|
||||
values = [-17, -18, -18]
|
||||
values: list[float] = [-17, -18, -18]
|
||||
values[position] = bad
|
||||
assert not debt_is_admitted(*values, STATE_ALIGNED_DEBT)
|
||||
assert not debt_is_admitted(values[0], values[1], values[2], STATE_ALIGNED_DEBT)
|
||||
rejects(lambda: clipping_quality('unknown'))
|
||||
rejects(lambda: event_clipping_quality({'event_version': 2, 'event_clipping_quality': STATE_ALIGNED_DEBT}))
|
||||
rejects(lambda: event_clipping_quality({'event_version': 3, 'event_clipping_quality': 'unknown'}))
|
||||
|
||||
@@ -50,6 +50,7 @@ def test_real_driver_measured_unwind_relief_is_weaker_and_final_only(platform):
|
||||
EncoderBranch(quality_contract=STATE_ALIGNED_DEBT).step(event, candidate)
|
||||
branch = EncoderBranch(quality_contract=STATE_ALIGNED_RELIEF)
|
||||
assert branch.step(event, candidate) is not None
|
||||
assert branch.previous is not None
|
||||
assert 0 < branch.previous <= witness.final_raw
|
||||
# Actual recorded no-clipping status cannot be bypassed by relief.
|
||||
with pytest.raises(ValueError, match='limiter/support'):
|
||||
|
||||
@@ -20,7 +20,7 @@ from openpilot.starpilot.flm.tests.test_event_fd_boundary import engine
|
||||
def test_exact_unwind_witnesses_and_partial_order_negatives():
|
||||
assert native_relief_is_admitted(-76, -76, -75, -80, -79, -79, 4, 3, 4)
|
||||
assert native_relief_is_admitted(-19, -19, -18, -28, -28, -28, 9, 9, 10)
|
||||
valid = [-76, -76, -75, -80, -79, -79, 4, 3, 4]
|
||||
valid: list[float] = [-76, -76, -75, -80, -79, -79, 4, 3, 4]
|
||||
for index, value in ((2, -77), (2, 0), (2, 1), (5, -81), (5, 79), (6, -4), (7, -3), (8, -4), (8, 0), (1, float('nan'))):
|
||||
changed = valid.copy()
|
||||
changed[index] = value
|
||||
|
||||
@@ -45,10 +45,13 @@ def payload():
|
||||
def plan_for(restored):
|
||||
model = fitted_model()
|
||||
base = primitive_plan(model)
|
||||
values = {**asdict(base), 'version': RESTORE_ACTIVATION_VERSION,
|
||||
'baseline_revision': restored.selected_baseline_revision, 'source_epoch': restored.source_epoch,
|
||||
'cutoff_ns': 2_005_000_000, 'windows': ((2_000_000_000, 2_020_000_000, 2_010_000_000, 2_010_000_000),)}
|
||||
return WholeMapRestorePlan(**values, restore_envelope_sha256=restore_envelope_digest(restored)).validate()
|
||||
return WholeMapRestorePlan(
|
||||
version=RESTORE_ACTIVATION_VERSION, identity=base.identity,
|
||||
baseline_revision=restored.selected_baseline_revision, source_epoch=restored.source_epoch, physical_context=base.physical_context,
|
||||
calibration_train_blocks=base.calibration_train_blocks, calibration_heldout_blocks=base.calibration_heldout_blocks,
|
||||
cutoff_ns=2_005_000_000, stage_blocks=base.stage_blocks,
|
||||
windows=((2_000_000_000, 2_020_000_000, 2_010_000_000, 2_010_000_000),),
|
||||
restore_envelope_sha256=restore_envelope_digest(restored)).validate()
|
||||
|
||||
|
||||
def test_restore_only_payload_has_no_shadow_or_operation_fallback():
|
||||
|
||||
@@ -20,11 +20,11 @@ def test_response_after_command_is_fresh_only_at_a_trusted_later_decision():
|
||||
|
||||
@pytest.mark.parametrize('index', (0, 1, 2))
|
||||
def test_each_original_stream_must_be_nonfuture_and_within_100ms(index):
|
||||
valid = [NOW - 2_000_000, NOW - 1_000_000, NOW]
|
||||
valid: list[int | float] = [NOW - 2_000_000, NOW - 1_000_000, NOW]
|
||||
for stamp in (NOW + 1, NOW - 101_000_000, 0, -1, True, float(NOW)):
|
||||
values = valid.copy()
|
||||
values[index] = stamp
|
||||
assert not _fresh_current_physical(*values, NOW)
|
||||
assert not _fresh_current_physical(values[0], values[1], values[2], NOW)
|
||||
assert not _fresh_current_physical(NOW, NOW - 1, NOW, NOW)
|
||||
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Exact original-frame join; no synthetic replay qualification."""
|
||||
from types import SimpleNamespace
|
||||
from typing import TypedDict
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
@@ -8,6 +9,20 @@ from openpilot.starpilot.flm.event_raw_fit import _control_frame_index, fit_even
|
||||
from openpilot.starpilot.flm.tests.test_event_activation import fitted_model
|
||||
|
||||
|
||||
class StageSample(TypedDict):
|
||||
control_ns: int
|
||||
apply_ns: int
|
||||
mono_ns: int
|
||||
control: dict[str, int]
|
||||
output: dict[str, int]
|
||||
physical: dict[str, float]
|
||||
|
||||
|
||||
class StageEvent(TypedDict):
|
||||
block_id: str
|
||||
samples: tuple[StageSample, ...]
|
||||
|
||||
|
||||
def test_index_preserves_original_frame_identity_and_missing_timestamp():
|
||||
first = SimpleNamespace(control_ns=10, desired_physical_accel=.02)
|
||||
second = SimpleNamespace(control_ns=20, desired_physical_accel=.03)
|
||||
@@ -29,7 +44,7 @@ def test_actual_fit_missing_stage_frame_is_fail_closed():
|
||||
# Isolate the stage join after its independently checked upstream operations.
|
||||
# The model comes from the real measured-row fitter, never a qualification flag.
|
||||
model = fitted_model()
|
||||
events = tuple({'block_id': str(index), 'samples': ({'control_ns': index * 1_000_000_000,
|
||||
events: tuple[StageEvent, ...] = tuple({'block_id': str(index), 'samples': ({'control_ns': index * 1_000_000_000,
|
||||
'apply_ns': index * 1_000_000_000 + 1, 'mono_ns': index * 1_000_000_000 + 2,
|
||||
'control': {'stamp': index * 1_000_000_000}, 'output': {'stamp': index * 1_000_000_000 + 1},
|
||||
'physical': {'lateral_accel': .02}},)} for index in range(1, 17))
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
"""Model noise remains ineligible even for explicitly retained controller chatter."""
|
||||
from dataclasses import asdict
|
||||
from dataclasses import replace
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -35,4 +35,4 @@ def test_exact_canonical_objective_only_retains_expected_chatter(group):
|
||||
def test_nonfriction_coordinates_cannot_claim_friction_contract():
|
||||
objective = EffectObjective('friction.left', GROUPS['friction.left'], 'a' * 64, 'b' * 64)
|
||||
with pytest.raises(ValueError, match='canonical event group'):
|
||||
EffectObjective(**{**asdict(objective), 'names': GROUPS['ff.left']})
|
||||
replace(objective, names=GROUPS['ff.left'])
|
||||
|
||||
@@ -17,7 +17,7 @@ from openpilot.starpilot.flm.learner_factory import TorqueCollectionAdapter, RAW
|
||||
from openpilot.starpilot.flm.runtime_adapter_registry import baseline_values, baseline_revision, selected_descriptor, selected_identity
|
||||
from openpilot.starpilot.flm.tests.native_controls_fixture import native_frame
|
||||
from openpilot.starpilot.flm.tests.test_registered_controls_global import controls_for, published_cp
|
||||
from openpilot.starpilot.lateral.controller_selection import DOCUMENT_KEY as SELECTION_KEY, replace_mode
|
||||
from openpilot.starpilot.lateral.controller_selection import DOCUMENT_KEY as SELECTION_KEY, ControllerMode, replace_mode
|
||||
from openpilot.starpilot.lateral.learning_preferences import KEY, replace_preference
|
||||
from openpilot.starpilot.lateral.torque_settings import DOCUMENT_KEY, FieldChoice, PlatformProfile, serialize_document
|
||||
|
||||
@@ -104,14 +104,15 @@ def test_off_keeps_historical_controller_output_despite_qualified_native_packet(
|
||||
@contextmanager
|
||||
def manual_ioniq():
|
||||
cp = cp_for(HYUNDAI.HYUNDAI_IONIQ_6)
|
||||
basis = tuple(float(getattr(cp.lateralTuning.torque, name)) for name in ('latAccelFactor', 'latAccelOffset', 'friction'))
|
||||
torque = cp.lateralTuning.torque
|
||||
basis = (float(torque.latAccelFactor), float(torque.latAccelOffset), float(torque.friction))
|
||||
document = serialize_document({str(cp.carFingerprint): PlatformProfile(basis, FieldChoice('custom', basis[0] * .95), FieldChoice())})
|
||||
with OpenpilotPrefix(), mock.patch.dict(os.environ, {'REPLAY': '1', 'SIMULATION': '0', 'TORQUE_REPLAY_RUNTIME': '0',
|
||||
'AOL_REPLAY_RUNTIME': '0', 'LANE_CENTERING_REPLAY_RUNTIME': '0'}), mock.patch('openpilot.selfdrive.controls.controlsd.messaging.PubMaster'):
|
||||
params = Params()
|
||||
params.put('CarParams', published_cp(cp, params).to_bytes(), block=True)
|
||||
params.put(KEY, replace_preference(None, cp, True), block=True)
|
||||
params.put(SELECTION_KEY, json.loads(replace_mode(None, cp, 'starpilot')), block=True)
|
||||
params.put(SELECTION_KEY, json.loads(replace_mode(None, cp, ControllerMode.STARPILOT)), block=True)
|
||||
params.put(DOCUMENT_KEY, json.loads(document), block=True)
|
||||
yield Controls(), document, basis
|
||||
|
||||
|
||||
@@ -102,7 +102,10 @@ def test_honda_normalized_pre_lookup_comparison_and_modified_filter_shadow_is_pu
|
||||
assert observed_branch.civic == prior_civic # paired evaluation never advances filter
|
||||
observed_shapes.append(shape)
|
||||
return _shaped_input(row, request, shape)
|
||||
candidate_shape = None if not modified else before + tuple(witness.shape[4:])
|
||||
candidate_shape = None
|
||||
if modified:
|
||||
assert before is not None
|
||||
candidate_shape = before + tuple(witness.shape[4:])
|
||||
expected_source = _shaped_input(witness, witness.requested_torque, candidate_shape)
|
||||
params = dict(witness.limits)
|
||||
expected = rate_limit(expected_source, branch.previous,
|
||||
|
||||
@@ -23,12 +23,13 @@ def incoming(identity, *, frame=0):
|
||||
if identity != 'VOLVO_V40':
|
||||
return sources(identity, frame=frame)
|
||||
packer = CANPacker('volvo_v40_2017_pt')
|
||||
return [packer.make_can_msg(name, bus, values) for name, bus, values in [
|
||||
messages: list[tuple[str, int, dict[str, float]]] = [
|
||||
('VehicleSpeed1', 0, {'VehicleSpeed': 72}), ('CCButtons', 0, {}),
|
||||
('PSCM1', 0, {'SteeringAngleServo': 12.5, 'LKATorque': 10}),
|
||||
('PedalandBrake', 0, {}), ('TCM0', 0, {'GearShifter': 3}), ('ACC', 0, {'SpeedTargetACC': 100}),
|
||||
('MiscCarInfo', 0, {}), ('FSM0', 2, {'ACCStatusOnOff': 1, 'ACCStatusActive': 1}), ('FSM1', 2, {}),
|
||||
]]
|
||||
]
|
||||
return [packer.make_can_msg(name, bus, values) for name, bus, values in messages]
|
||||
|
||||
|
||||
class TestVolvoActualStartup(unittest.TestCase):
|
||||
|
||||
@@ -184,7 +184,7 @@ def test_actual_selfdrived_fault_clear_does_not_reengage_without_pcm_edge(identi
|
||||
if lib.safety_config_valid() and fresh_panda:
|
||||
break
|
||||
else:
|
||||
pytest.fail('Native RX and a fresh healthy Panda receipt did not recover within one source period')
|
||||
pytest.fail(reason='Native RX and a fresh healthy Panda receipt did not recover within one source period')
|
||||
assert not sd.enabled
|
||||
_, command = cycle(True)
|
||||
assert sd.enabled and sd.active and command.latActive
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
{"protected_inventory_sha256": "93618a5ca359682ede714b04c6e50d87c0cf63a50e5f5ab05d1d1729cc38a08d", "schema_version": 1}
|
||||
{"protected_inventory_sha256": "961f82df6e41b06fefdea7ced62e8e635225d545d9a3f3419b6e96f4d410607c", "schema_version": 1}
|
||||
|
||||
Reference in New Issue
Block a user