From 598074c192d8019d06a7a7a6e2b34119c30ace20 Mon Sep 17 00:00:00 2001 From: Jason Young <46612682+jyoung8607@users.noreply.github.com> Date: Thu, 20 Feb 2020 16:57:07 -0500 Subject: [PATCH] Volkswagen safety updates: Phase 2 (#445) * CRC and counter checks, standstill/brake checks * Clean up a tsk_06 that snuck through * Be consistent about how we call _msg_esp_05 * Reduce scope: haunted by the ghost of MISRA future * Improved check/test for in-motion braking * MISRA styling fix --- board/safety.h | 15 +++ board/safety/safety_volkswagen.h | 82 +++++++++++++- board/safety_declarations.h | 1 + requirements.txt | 1 + tests/safety/libpandasafety_py.py | 2 + tests/safety/test.c | 11 ++ tests/safety/test_volkswagen_mqb.py | 161 +++++++++++++++++++++++++++- 7 files changed, 266 insertions(+), 7 deletions(-) diff --git a/board/safety.h b/board/safety.h index b4c94ef..e13021f 100644 --- a/board/safety.h +++ b/board/safety.h @@ -57,6 +57,21 @@ int safety_fwd_hook(int bus_num, CAN_FIFOMailBox_TypeDef *to_fwd) { return current_hooks->fwd(bus_num, to_fwd); } +// Given a CRC-8 poly, generate a static lookup table to use with a fast CRC-8 +// algorithm. Called at init time for safety modes using CRC-8. +void gen_crc_lookup_table(uint8_t poly, uint8_t crc_lut[]) { + for (int i = 0; i < 256; i++) { + uint8_t crc = i; + for (int j = 0; j < 8; j++) { + if ((crc & 0x80U) != 0U) + crc = (uint8_t)((crc << 1) ^ poly); + else + crc <<= 1; + } + crc_lut[i] = crc; + } +} + bool msg_allowed(int addr, int bus, const AddrBus addr_list[], int len) { bool allowed = false; for (int i = 0; i < len; i++) { diff --git a/board/safety/safety_volkswagen.h b/board/safety/safety_volkswagen.h index af15a9f..20b4d56 100644 --- a/board/safety/safety_volkswagen.h +++ b/board/safety/safety_volkswagen.h @@ -8,7 +8,9 @@ const int VOLKSWAGEN_DRIVER_TORQUE_ALLOWANCE = 80; const int VOLKSWAGEN_DRIVER_TORQUE_FACTOR = 3; // Safety-relevant CAN messages for the Volkswagen MQB platform +#define MSG_ESP_19 0x0B2 // RX from ABS, for wheel speeds #define MSG_EPS_01 0x09F // RX from EPS, for driver steering torque +#define MSG_ESP_05 0x106 // RX from ABS, for brake switch state #define MSG_MOTOR_20 0x121 // RX from ECU, for driver throttle input #define MSG_ACC_06 0x122 // RX from ACC radar, for status and engagement #define MSG_HCA_01 0x126 // TX by OP, Heading Control Assist steering torque @@ -19,11 +21,12 @@ const int VOLKSWAGEN_DRIVER_TORQUE_FACTOR = 3; const AddrBus VOLKSWAGEN_MQB_TX_MSGS[] = {{MSG_HCA_01, 0}, {MSG_GRA_ACC_01, 0}, {MSG_GRA_ACC_01, 2}, {MSG_LDW_02, 0}}; const int VOLKSWAGEN_MQB_TX_MSGS_LEN = sizeof(VOLKSWAGEN_MQB_TX_MSGS) / sizeof(VOLKSWAGEN_MQB_TX_MSGS[0]); -// TODO: do checksum and counter checks AddrCheckStruct volkswagen_mqb_rx_checks[] = { - {.addr = {MSG_EPS_01}, .bus = 0, .expected_timestep = 10000U}, - {.addr = {MSG_ACC_06}, .bus = 0, .expected_timestep = 20000U}, - {.addr = {MSG_MOTOR_20}, .bus = 0, .expected_timestep = 20000U}, + {.addr = {MSG_ESP_19}, .bus = 0, .check_checksum = false, .max_counter = 0U, .expected_timestep = 10000U}, + {.addr = {MSG_EPS_01}, .bus = 0, .check_checksum = true, .max_counter = 15U, .expected_timestep = 10000U}, + {.addr = {MSG_ESP_05}, .bus = 0, .check_checksum = true, .max_counter = 15U, .expected_timestep = 20000U}, + {.addr = {MSG_MOTOR_20}, .bus = 0, .check_checksum = true, .max_counter = 15U, .expected_timestep = 20000U}, + {.addr = {MSG_ACC_06}, .bus = 0, .check_checksum = true, .max_counter = 15U, .expected_timestep = 20000U}, }; const int VOLKSWAGEN_MQB_RX_CHECKS_LEN = sizeof(volkswagen_mqb_rx_checks) / sizeof(volkswagen_mqb_rx_checks[0]); @@ -32,9 +35,56 @@ struct sample_t volkswagen_torque_driver; // Last few driver torques measured int volkswagen_rt_torque_last = 0; int volkswagen_desired_torque_last = 0; uint32_t volkswagen_ts_last = 0; +bool volkswagen_moving = false; +bool volkswagen_brake_pressed_prev = false; int volkswagen_gas_prev = 0; int volkswagen_torque_msg = 0; int volkswagen_lane_msg = 0; +uint8_t volkswagen_crc8_lut_8h2f[256]; // Static lookup table for CRC8 poly 0x2F, aka 8H2F/AUTOSAR + + +static uint8_t volkswagen_get_checksum(CAN_FIFOMailBox_TypeDef *to_push) { + return (uint8_t)GET_BYTE(to_push, 0); +} + +static uint8_t volkswagen_get_counter(CAN_FIFOMailBox_TypeDef *to_push) { + return (uint8_t)GET_BYTE(to_push, 1) & 0xFU; +} + +static uint8_t volkswagen_mqb_compute_crc(CAN_FIFOMailBox_TypeDef *to_push) { + int addr = GET_ADDR(to_push); + int len = GET_LEN(to_push); + + // This is CRC-8H2F/AUTOSAR with a twist. See the OpenDBC implementation + // of this algorithm for a version with explanatory comments. + + uint8_t crc = 0xFFU; + for (int i = 1; i < len; i++) { + crc ^= (uint8_t)GET_BYTE(to_push, i); + crc = volkswagen_crc8_lut_8h2f[crc]; + } + + uint8_t counter = volkswagen_get_counter(to_push); + switch(addr) { + case MSG_EPS_01: + crc ^= (uint8_t[]){0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5,0xF5}[counter]; + break; + case MSG_ESP_05: + crc ^= (uint8_t[]){0x07,0x07,0x07,0x07,0x07,0x07,0x07,0x07,0x07,0x07,0x07,0x07,0x07,0x07,0x07,0x07}[counter]; + break; + case MSG_MOTOR_20: + crc ^= (uint8_t[]){0xE9,0x65,0xAE,0x6B,0x7B,0x35,0xE5,0x5F,0x4E,0xC7,0x86,0xA2,0xBB,0xDD,0xEB,0xB4}[counter]; + break; + case MSG_ACC_06: + crc ^= (uint8_t[]){0x37,0x7D,0xF3,0xA9,0x18,0x46,0x6D,0x4D,0x3D,0x71,0x92,0x9C,0xE5,0x32,0x10,0xB9}[counter]; + break; + default: // Undefined CAN message, CRC check expected to fail + break; + } + crc = volkswagen_crc8_lut_8h2f[crc]; + + return crc ^ 0xFFU; +} static void volkswagen_mqb_init(int16_t param) { UNUSED(param); @@ -43,17 +93,29 @@ static void volkswagen_mqb_init(int16_t param) { relay_malfunction = false; volkswagen_torque_msg = MSG_HCA_01; volkswagen_lane_msg = MSG_LDW_02; + gen_crc_lookup_table(0x2F, volkswagen_crc8_lut_8h2f); } static int volkswagen_mqb_rx_hook(CAN_FIFOMailBox_TypeDef *to_push) { bool valid = addr_safety_check(to_push, volkswagen_mqb_rx_checks, VOLKSWAGEN_MQB_RX_CHECKS_LEN, - NULL, NULL, NULL); + volkswagen_get_checksum, volkswagen_mqb_compute_crc, volkswagen_get_counter); if (valid) { int bus = GET_BUS(to_push); int addr = GET_ADDR(to_push); + // Update in-motion state by sampling front wheel speeds + // Signal: ESP_19.ESP_VL_Radgeschw_02 (front left) in scaled km/h + // Signal: ESP_19.ESP_VR_Radgeschw_02 (front right) in scaled km/h + if ((bus == 0) && (addr == MSG_ESP_19)) { + int wheel_speed_fl = GET_BYTE(to_push, 4) | (GET_BYTE(to_push, 5) << 8); + int wheel_speed_fr = GET_BYTE(to_push, 6) | (GET_BYTE(to_push, 7) << 8); + // Check for average front speed in excess of 0.3m/s, 1.08km/h + // DBC speed scale 0.0075: 0.3m/s = 144, sum both wheels to compare + volkswagen_moving = (wheel_speed_fl + wheel_speed_fr) > 288; + } + // Update driver input torque samples // Signal: EPS_01.Driver_Strain (absolute torque) // Signal: EPS_01.Driver_Strain_VZ (direction) @@ -83,6 +145,16 @@ static int volkswagen_mqb_rx_hook(CAN_FIFOMailBox_TypeDef *to_push) { volkswagen_gas_prev = gas; } + // Exit controls on rising edge of brake press + // Signal: ESP_05.ESP_Fahrer_bremst + if ((bus == 0) && (addr == MSG_ESP_05)) { + bool brake_pressed = (GET_BYTE(to_push, 3) & 0x4) >> 2; + if (brake_pressed && (!(volkswagen_brake_pressed_prev) || volkswagen_moving)) { + controls_allowed = 0; + } + volkswagen_brake_pressed_prev = brake_pressed; + } + // If there are HCA messages on bus 0 not sent by OP, there's a relay problem if ((safety_mode_cnt > RELAY_TRNS_TIMEOUT) && (bus == 0) && (addr == MSG_HCA_01)) { relay_malfunction = true; diff --git a/board/safety_declarations.h b/board/safety_declarations.h index 1a6a8dd..1e1aa5a 100644 --- a/board/safety_declarations.h +++ b/board/safety_declarations.h @@ -49,6 +49,7 @@ bool driver_limit_check(int val, int val_last, struct sample_t *val_driver, const int MAX_ALLOWANCE, const int DRIVER_FACTOR); bool rt_rate_limit_check(int val, int val_last, const int MAX_RT_DELTA); float interpolate(struct lookup_t xy, float x); +void gen_crc_lookup_table(uint8_t poly, uint8_t crc_lut[]); bool msg_allowed(int addr, int bus, const AddrBus addr_list[], int len); int get_addr_check_index(CAN_FIFOMailBox_TypeDef *to_push, AddrCheckStruct addr_list[], const int len); void update_counter(AddrCheckStruct addr_list[], int index, uint8_t counter); diff --git a/requirements.txt b/requirements.txt index 5ceb3cf..fb01edc 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,3 +9,4 @@ requests flake8==3.7.9 pylint==2.4.3 cffi==1.11.4 +crcmod diff --git a/tests/safety/libpandasafety_py.py b/tests/safety/libpandasafety_py.py index 4dfaef1..82ae5b1 100644 --- a/tests/safety/libpandasafety_py.py +++ b/tests/safety/libpandasafety_py.py @@ -94,6 +94,8 @@ void init_tests_volkswagen(void); int get_volkswagen_gas_prev(void); int get_volkswagen_torque_driver_min(void); int get_volkswagen_torque_driver_max(void); +bool get_volkswagen_moving(void); +bool get_volkswagen_brake_pressed_prev(void); void set_volkswagen_desired_torque_last(int t); void set_volkswagen_rt_torque_last(int t); void set_volkswagen_torque_driver(int min, int max); diff --git a/tests/safety/test.c b/tests/safety/test.c index 85b87ff..5c405d8 100644 --- a/tests/safety/test.c +++ b/tests/safety/test.c @@ -232,6 +232,14 @@ void set_volkswagen_desired_torque_last(int t){ volkswagen_desired_torque_last = t; } +int get_volkswagen_moving(void){ + return volkswagen_moving; +} + +int get_volkswagen_brake_pressed_prev(void){ + return volkswagen_brake_pressed_prev; +} + int get_volkswagen_gas_prev(void){ return volkswagen_gas_prev; } @@ -334,6 +342,9 @@ void init_tests_subaru(void){ void init_tests_volkswagen(void){ init_tests(); + volkswagen_moving = false; + volkswagen_brake_pressed_prev = false; + volkswagen_gas_prev = 0; volkswagen_torque_driver.min = 0; volkswagen_torque_driver.max = 0; volkswagen_desired_torque_last = 0; diff --git a/tests/safety/test_volkswagen_mqb.py b/tests/safety/test_volkswagen_mqb.py index 9f5b461..8874c59 100644 --- a/tests/safety/test_volkswagen_mqb.py +++ b/tests/safety/test_volkswagen_mqb.py @@ -1,10 +1,11 @@ #!/usr/bin/env python3 import unittest import numpy as np +import crcmod from panda import Panda from panda.tests.safety import libpandasafety_py from panda.tests.safety.common import test_relay_malfunction, make_msg, \ - test_manually_enable_controls_allowed, test_spam_can_buses + test_manually_enable_controls_allowed, test_spam_can_buses, MAX_WRONG_COUNTERS MAX_RATE_UP = 4 MAX_RATE_DOWN = 10 @@ -15,9 +16,11 @@ RT_INTERVAL = 250000 DRIVER_TORQUE_ALLOWANCE = 80 DRIVER_TORQUE_FACTOR = 3 +MSG_ESP_19 = 0xB2 # RX from ABS, for wheel speeds MSG_EPS_01 = 0x9F # RX from EPS, for driver steering torque -MSG_ACC_06 = 0x122 # RX from ACC radar, for status and engagement +MSG_ESP_05 = 0x106 # RX from ABS, for brake light state MSG_MOTOR_20 = 0x121 # RX from ECU, for driver throttle input +MSG_ACC_06 = 0x122 # RX from ACC radar, for status and engagement MSG_HCA_01 = 0x126 # TX by OP, Heading Control Assist steering torque MSG_GRA_ACC_01 = 0x12B # TX by OP, ACC control buttons for cancel/resume MSG_LDW_02 = 0x397 # TX by OP, Lane line recognition and text alerts @@ -31,17 +34,66 @@ def sign(a): else: return -1 +# Python crcmod works differently somehow from every other CRC calculator. The +# implied leading 1 on the polynomial isn't a problem, but to get the right +# result for CRC-8H2F/AUTOSAR, we have to feed it initCrc 0x00 instead of 0xFF. +volkswagen_crc_8h2f = crcmod.mkCrcFun(0x12F, initCrc=0x00, rev=False, xorOut=0xFF) + +def volkswagen_mqb_crc(msg, addr, len_msg): + # This is CRC-8H2F/AUTOSAR with a twist. See the OpenDBC implementation of + # this algorithm for a version with explanatory comments. + msg_bytes = msg.RDLR.to_bytes(4, 'little') + msg.RDHR.to_bytes(4, 'little') + counter = (msg.RDLR & 0xF00) >> 8 + if addr == MSG_EPS_01: + magic_pad = b'\xF5\xF5\xF5\xF5\xF5\xF5\xF5\xF5\xF5\xF5\xF5\xF5\xF5\xF5\xF5\xF5'[counter] + elif addr == MSG_ESP_05: + magic_pad = b'\x07\x07\x07\x07\x07\x07\x07\x07\x07\x07\x07\x07\x07\x07\x07\x07'[counter] + elif addr == MSG_MOTOR_20: + magic_pad = b'\xE9\x65\xAE\x6B\x7B\x35\xE5\x5F\x4E\xC7\x86\xA2\xBB\xDD\xEB\xB4'[counter] + elif addr == MSG_ACC_06: + magic_pad = b'\x37\x7D\xF3\xA9\x18\x46\x6D\x4D\x3D\x71\x92\x9C\xE5\x32\x10\xB9'[counter] + elif addr == MSG_HCA_01: + magic_pad = b'\xDA\xDA\xDA\xDA\xDA\xDA\xDA\xDA\xDA\xDA\xDA\xDA\xDA\xDA\xDA\xDA'[counter] + elif addr == MSG_GRA_ACC_01: + magic_pad = b'\x6A\x38\xB4\x27\x22\xEF\xE1\xBB\xF8\x80\x84\x49\xC7\x9E\x1E\x2B'[counter] + else: + magic_pad = None + return volkswagen_crc_8h2f(msg_bytes[1:len_msg] + magic_pad.to_bytes(1, 'little')) + class TestVolkswagenMqbSafety(unittest.TestCase): @classmethod def setUp(cls): cls.safety = libpandasafety_py.libpandasafety cls.safety.set_safety_hooks(Panda.SAFETY_VOLKSWAGEN_MQB, 0) cls.safety.init_tests_volkswagen() + cls.cnt_eps_01 = 0 + cls.cnt_esp_05 = 0 + cls.cnt_motor_20 = 0 + cls.cnt_acc_06 = 0 + cls.cnt_hca_01 = 0 + cls.cnt_gra_acc_01 = 0 def _set_prev_torque(self, t): self.safety.set_volkswagen_desired_torque_last(t) self.safety.set_volkswagen_rt_torque_last(t) + # Wheel speeds + def _esp_19_msg(self, speed): + wheel_speed_scaled = int(speed / 0.0075) + to_send = make_msg(0, MSG_ESP_19) + to_send[0].RDLR = wheel_speed_scaled | (wheel_speed_scaled << 16) + to_send[0].RDHR = wheel_speed_scaled | (wheel_speed_scaled << 16) + return to_send + + # Brake light switch + def _esp_05_msg(self, brake): + to_send = make_msg(0, MSG_ESP_05) + to_send[0].RDLR = (0x1 << 26) if brake else 0 + to_send[0].RDLR |= (self.cnt_esp_05 % 16) << 8 + to_send[0].RDLR |= volkswagen_mqb_crc(to_send[0], MSG_ESP_05, 8) + self.cnt_esp_05 += 1 + return to_send + # Driver steering input torque def _eps_01_msg(self, torque): to_send = make_msg(0, MSG_EPS_01) @@ -49,6 +101,9 @@ class TestVolkswagenMqbSafety(unittest.TestCase): to_send[0].RDHR = ((t & 0x1FFF) << 8) if torque < 0: to_send[0].RDHR |= 0x1 << 23 + to_send[0].RDLR |= (self.cnt_eps_01 % 16) << 8 + to_send[0].RDLR |= volkswagen_mqb_crc(to_send[0], MSG_EPS_01, 8) + self.cnt_eps_01 += 1 return to_send # openpilot steering output torque @@ -58,24 +113,36 @@ class TestVolkswagenMqbSafety(unittest.TestCase): to_send[0].RDLR = (t & 0xFFF) << 16 if torque < 0: to_send[0].RDLR |= 0x1 << 31 + to_send[0].RDLR |= (self.cnt_hca_01 % 16) << 8 + to_send[0].RDLR |= volkswagen_mqb_crc(to_send[0], MSG_HCA_01, 8) + self.cnt_hca_01 += 1 return to_send # ACC engagement status def _acc_06_msg(self, status): to_send = make_msg(0, MSG_ACC_06) to_send[0].RDHR = (status & 0x7) << 28 + to_send[0].RDLR |= (self.cnt_acc_06 % 16) << 8 + to_send[0].RDLR |= volkswagen_mqb_crc(to_send[0], MSG_ACC_06, 8) + self.cnt_acc_06 += 1 return to_send # Driver throttle input def _motor_20_msg(self, gas): to_send = make_msg(0, MSG_MOTOR_20) to_send[0].RDLR = (gas & 0xFF) << 12 + to_send[0].RDLR |= (self.cnt_motor_20 % 16) << 8 + to_send[0].RDLR |= volkswagen_mqb_crc(to_send[0], MSG_MOTOR_20, 8) + self.cnt_motor_20 += 1 return to_send # Cruise control buttons def _gra_acc_01_msg(self, bit): to_send = make_msg(2, MSG_GRA_ACC_01) to_send[0].RDLR = 1 << bit + to_send[0].RDLR |= (self.cnt_gra_acc_01 % 16) << 8 + to_send[0].RDLR |= volkswagen_mqb_crc(to_send[0], MSG_GRA_ACC_01, 8) + self.cnt_gra_acc_01 += 1 return to_send def test_spam_can_buses(self): @@ -102,6 +169,48 @@ class TestVolkswagenMqbSafety(unittest.TestCase): self.safety.safety_rx_hook(self._acc_06_msg(1)) self.assertFalse(self.safety.get_controls_allowed()) + def test_sample_speed(self): + # Stationary + self.safety.safety_rx_hook(self._esp_19_msg(0)) + self.assertEqual(0, self.safety.get_volkswagen_moving()) + # 1 km/h, just under 0.3 m/s safety grace threshold + self.safety.safety_rx_hook(self._esp_19_msg(1)) + self.assertEqual(0, self.safety.get_volkswagen_moving()) + # 2 km/h, just over 0.3 m/s safety grace threshold + self.safety.safety_rx_hook(self._esp_19_msg(2)) + self.assertEqual(1, self.safety.get_volkswagen_moving()) + # 144 km/h, openpilot V_CRUISE_MAX + self.safety.safety_rx_hook(self._esp_19_msg(144)) + self.assertEqual(1, self.safety.get_volkswagen_moving()) + + def test_prev_brake(self): + self.assertFalse(self.safety.get_volkswagen_brake_pressed_prev()) + self.safety.safety_rx_hook(self._esp_05_msg(True)) + self.assertTrue(self.safety.get_volkswagen_brake_pressed_prev()) + + def test_disengage_on_brake(self): + self.safety.set_controls_allowed(1) + self.safety.safety_rx_hook(self._esp_05_msg(True)) + self.assertFalse(self.safety.get_controls_allowed()) + + def test_allow_brake_at_zero_speed(self): + # Brake was already pressed + self.safety.safety_rx_hook(self._esp_05_msg(True)) + self.safety.set_controls_allowed(1) + + self.safety.safety_rx_hook(self._esp_05_msg(True)) + self.assertTrue(self.safety.get_controls_allowed()) + self.safety.safety_rx_hook(self._esp_05_msg(False)) # reset no brakes + + def test_not_allow_brake_when_moving(self): + # Brake was already pressed + self.safety.safety_rx_hook(self._esp_05_msg(True)) + self.safety.safety_rx_hook(self._esp_19_msg(100)) + self.safety.set_controls_allowed(1) + + self.safety.safety_rx_hook(self._esp_05_msg(True)) + self.assertFalse(self.safety.get_controls_allowed()) + def test_disengage_on_gas(self): self.safety.safety_rx_hook(self._motor_20_msg(0)) self.safety.set_controls_allowed(True) @@ -236,6 +345,54 @@ class TestVolkswagenMqbSafety(unittest.TestCase): self.assertEqual(0, self.safety.get_volkswagen_torque_driver_max()) self.assertEqual(0, self.safety.get_volkswagen_torque_driver_min()) + def test_rx_hook(self): + # checksum checks + # TODO: Would be ideal to check ESP_19 as well, but it has no checksum + # or counter, and I'm not sure if we can easily validate Panda's simple + # temporal reception-rate check here. + for msg in [MSG_EPS_01, MSG_ESP_05, MSG_MOTOR_20, MSG_ACC_06]: + self.safety.set_controls_allowed(1) + if msg == MSG_EPS_01: + to_push = self._eps_01_msg(0) + if msg == MSG_ESP_05: + to_push = self._esp_05_msg(False) + if msg == MSG_MOTOR_20: + to_push = self._motor_20_msg(0) + if msg == MSG_ACC_06: + to_push = self._acc_06_msg(3) + self.assertTrue(self.safety.safety_rx_hook(to_push)) + to_push[0].RDHR ^= 0xFF + self.assertFalse(self.safety.safety_rx_hook(to_push)) + self.assertFalse(self.safety.get_controls_allowed()) + + # counter + # reset wrong_counters to zero by sending valid messages + for i in range(MAX_WRONG_COUNTERS + 1): + self.cnt_eps_01 = 0 + self.cnt_esp_05 = 0 + self.cnt_motor_20 = 0 + self.cnt_acc_06 = 0 + if i < MAX_WRONG_COUNTERS: + self.safety.set_controls_allowed(1) + self.safety.safety_rx_hook(self._eps_01_msg(0)) + self.safety.safety_rx_hook(self._esp_05_msg(False)) + self.safety.safety_rx_hook(self._motor_20_msg(0)) + self.safety.safety_rx_hook(self._acc_06_msg(3)) + else: + self.assertFalse(self.safety.safety_rx_hook(self._eps_01_msg(0))) + self.assertFalse(self.safety.safety_rx_hook(self._esp_05_msg(False))) + self.assertFalse(self.safety.safety_rx_hook(self._motor_20_msg(0))) + self.assertFalse(self.safety.safety_rx_hook(self._acc_06_msg(3))) + self.assertFalse(self.safety.get_controls_allowed()) + + # restore counters for future tests with a couple of good messages + for i in range(2): + self.safety.set_controls_allowed(1) + self.safety.safety_rx_hook(self._eps_01_msg(0)) + self.safety.safety_rx_hook(self._esp_05_msg(False)) + self.safety.safety_rx_hook(self._motor_20_msg(0)) + self.safety.safety_rx_hook(self._acc_06_msg(3)) + self.assertTrue(self.safety.get_controls_allowed()) def test_fwd_hook(self): buss = list(range(0x0, 0x3))