Safety Test Refactor: Nissan (#510)

* start nissan

* packer

Author: adeebshihadeh

tests/safety/common.py

@@ -44,6 +44,7 @@ def make_can_msg_panda(self, name_or_addr, bus, values, counter=-1):
 class PandaSafetyTest(unittest.TestCase):
   TX_MSGS = None
   STANDSTILL_THRESHOLD = None
+  GAS_PRESSED_THRESHOLD = 0
   RELAY_MALFUNCTION_ADDR = None
   RELAY_MALFUNCTION_BUS = None
   FWD_BLACKLISTED_ADDRS = {} # {bus: [addr]}
@@ -119,11 +120,9 @@ def test_manually_enable_controls_allowed(self):
 
   def test_prev_gas(self):
     self.assertFalse(self.safety.get_gas_pressed_prev())
-    for pressed in [True, False]:
-      self._rx(self._gas_msg(not pressed))
-      self.assertEqual(not pressed, self.safety.get_gas_pressed_prev())
+    for pressed in [self.GAS_PRESSED_THRESHOLD+1, 0]:
       self._rx(self._gas_msg(pressed))
-      self.assertEqual(pressed, self.safety.get_gas_pressed_prev())
+      self.assertEqual(bool(pressed), self.safety.get_gas_pressed_prev())
 
   def test_allow_engage_with_gas_pressed(self):
     self._rx(self._gas_msg(1))
@@ -136,14 +135,14 @@ def test_allow_engage_with_gas_pressed(self):
   def test_disengage_on_gas(self):
     self._rx(self._gas_msg(0))
     self.safety.set_controls_allowed(True)
-    self._rx(self._gas_msg(1))
+    self._rx(self._gas_msg(self.GAS_PRESSED_THRESHOLD+1))
     self.assertFalse(self.safety.get_controls_allowed())
 
   def test_unsafe_mode_no_disengage_on_gas(self):
     self._rx(self._gas_msg(0))
     self.safety.set_controls_allowed(True)
     self.safety.set_unsafe_mode(UNSAFE_MODE.DISABLE_DISENGAGE_ON_GAS)
-    self._rx(self._gas_msg(1))
+    self._rx(self._gas_msg(self.GAS_PRESSED_THRESHOLD+1))
     self.assertTrue(self.safety.get_controls_allowed())
 
   def test_prev_brake(self):

tests/safety/test_nissan.py

@@ -3,41 +3,37 @@
 import numpy as np
 from panda import Panda
 from panda.tests.safety import libpandasafety_py
-from panda.tests.safety.common import StdTest, make_msg, UNSAFE_MODE
+import panda.tests.safety.common as common
+from panda.tests.safety.common import CANPackerPanda
 
 ANGLE_DELTA_BP = [0., 5., 15.]
 ANGLE_DELTA_V = [5., .8, .15]     # windup limit
 ANGLE_DELTA_VU = [5., 3.5, 0.4]   # unwind limit
 
-TX_MSGS = [[0x169, 0], [0x2b1, 0], [0x4cc, 0], [0x20b, 2], [0x280, 2]]
-
-def twos_comp(val, bits):
-  if val >= 0:
-    return val
-  else:
-    return (2**bits) + val
 
 def sign(a):
-  if a > 0:
-    return 1
-  else:
-    return -1
+  return 1 if a > 0 else -1
 
 
-class TestNissanSafety(unittest.TestCase):
-  @classmethod
-  def setUp(cls):
-    cls.safety = libpandasafety_py.libpandasafety
-    cls.safety.set_safety_hooks(Panda.SAFETY_NISSAN, 0)
-    cls.safety.init_tests_nissan()
+class TestNissanSafety(common.PandaSafetyTest):
 
-  def _angle_meas_msg(self, angle):
-    to_send = make_msg(0, 0x2)
-    angle = int(angle * -10)
-    t = twos_comp(angle, 16)
-    to_send[0].RDLR = t & 0xFFFF
+  TX_MSGS = [[0x169, 0], [0x2b1, 0], [0x4cc, 0], [0x20b, 2], [0x280, 2]]
+  STANDSTILL_THRESHOLD = 0
+  GAS_PRESSED_THRESHOLD = 1
+  RELAY_MALFUNCTION_ADDR = 0x169
+  RELAY_MALFUNCTION_BUS = 0
+  FWD_BLACKLISTED_ADDRS = {0: [0x280], 2: [0x169, 0x2b1, 0x4cc]}
+  FWD_BUS_LOOKUP = {0: 2, 2: 0}
+
+  def setUp(self):
+    self.packer = CANPackerPanda("nissan_x_trail_2017")
+    self.safety = libpandasafety_py.libpandasafety
+    self.safety.set_safety_hooks(Panda.SAFETY_NISSAN, 0)
+    self.safety.init_tests_nissan()
 
-    return to_send
+  def _angle_meas_msg(self, angle):
+    values = {"STEER_ANGLE": angle}
+    return self.packer.make_can_msg_panda("STEER_ANGLE_SENSOR", 0, values)
 
   def _set_prev_angle(self, t):
     t = int(t * -100)
@@ -47,47 +43,33 @@ def _angle_meas_msg_array(self, angle):
     for i in range(6):
       self.safety.safety_rx_hook(self._angle_meas_msg(angle))
 
-  def _lkas_state_msg(self, state):
-    to_send = make_msg(1, 0x30f)
-    to_send[0].RDHR = (state & 0x1) << 3
-
-    return to_send
+  def _pcm_status_msg(self, enabled):
+    values = {"CRUISE_ENABLED": enabled}
+    return self.packer.make_can_msg_panda("CRUISE_STATE", 2, values)
 
   def _lkas_control_msg(self, angle, state):
-    to_send = make_msg(0, 0x169)
-    angle = int((angle - 1310) * -100)
-    to_send[0].RDLR = ((angle & 0x3FC00) >> 10) | ((angle & 0x3FC) << 6) | ((angle & 0x3) << 16)
-    to_send[0].RDHR = ((state & 0x1) << 20)
-
-    return to_send
+    values = {"DESIRED_ANGLE": angle, "LKA_ACTIVE": state}
+    return self.packer.make_can_msg_panda("LKAS", 0, values)
 
   def _speed_msg(self, speed):
-    to_send = make_msg(0, 0x285)
-    speed = int(speed / 0.005 * 3.6)
-    to_send[0].RDLR = ((speed & 0xFF) << 24) | ((speed & 0xFF00) << 8)
-
-    return to_send
+    # TODO: why the 3.6? m/s to kph? not in dbc
+    values = {"WHEEL_SPEED_%s"%s: speed*3.6 for s in ["RR", "RL"]}
+    return self.packer.make_can_msg_panda("WHEEL_SPEEDS_REAR", 0, values)
 
   def _brake_msg(self, brake):
-    to_send = make_msg(1, 0x454)
-    to_send[0].RDLR = ((brake & 0x1) << 23)
-
-    return to_send
-
-  def _send_gas_cmd(self, gas):
-    to_send = make_msg(0, 0x15c)
-    to_send[0].RDHR = ((gas & 0x3fc) << 6) | ((gas & 0x3) << 22)
-
-    return to_send
-
-  def _acc_button_cmd(self, buttons):
-    to_send = make_msg(2, 0x20b)
-    to_send[0].RDLR = (buttons << 8)
+    values = {"USER_BRAKE_PRESSED": brake}
+    return self.packer.make_can_msg_panda("DOORS_LIGHTS", 1, values)
 
-    return to_send
+  def _gas_msg(self, gas):
+    values = {"GAS_PEDAL": gas}
+    return self.packer.make_can_msg_panda("GAS_PEDAL", 0, values)
 
-  def test_spam_can_buses(self):
-    StdTest.test_spam_can_buses(self, TX_MSGS)
+  def _acc_button_cmd(self, cancel=0, propilot=0, flw_dist=0, _set=0, res=0):
+    no_button = not any([cancel, propilot, flw_dist, _set, res])
+    values = {"CANCEL_BUTTON": cancel, "PROPILOT_BUTTON": propilot, \
+              "FOLLOW_DISTANCE_BUTTON": flw_dist, "SET_BUTTON": _set, \
+              "RES_BUTTON": res, "NO_BUTTON_PRESSED": no_button}
+    return self.packer.make_can_msg_panda("CRUISE_THROTTLE", 2, values)
 
   def test_angle_cmd_when_enabled(self):
     # when controls are allowed, angle cmd rate limit is enforced
@@ -145,65 +127,25 @@ def test_angle_cmd_when_disabled(self):
     self.assertFalse(self.safety.safety_tx_hook(self._lkas_control_msg(0, 1)))
     self.assertFalse(self.safety.get_controls_allowed())
 
-  def test_brake_disengage(self):
-    StdTest.test_allow_brake_at_zero_speed(self)
-    StdTest.test_not_allow_brake_when_moving(self, 0)
-
-  def test_gas_rising_edge(self):
-    self.safety.set_controls_allowed(1)
-    self.safety.safety_rx_hook(self._send_gas_cmd(100))
-    self.assertFalse(self.safety.get_controls_allowed())
-
-  def test_unsafe_mode_no_disengage_on_gas(self):
-    self.safety.safety_rx_hook(self._send_gas_cmd(0))
-    self.safety.set_controls_allowed(True)
-    self.safety.set_unsafe_mode(UNSAFE_MODE.DISABLE_DISENGAGE_ON_GAS)
-    self.safety.safety_rx_hook(self._send_gas_cmd(100))
-    self.assertTrue(self.safety.get_controls_allowed())
-    self.safety.set_unsafe_mode(UNSAFE_MODE.DEFAULT)
-
   def test_acc_buttons(self):
     self.safety.set_controls_allowed(1)
-    self.safety.safety_tx_hook(self._acc_button_cmd(0x2)) # Cancel button
+    self.safety.safety_tx_hook(self._acc_button_cmd(cancel=1))
     self.assertTrue(self.safety.get_controls_allowed())
-    self.safety.safety_tx_hook(self._acc_button_cmd(0x1)) # ProPilot button
+    self.safety.safety_tx_hook(self._acc_button_cmd(propilot=1))
     self.assertFalse(self.safety.get_controls_allowed())
     self.safety.set_controls_allowed(1)
-    self.safety.safety_tx_hook(self._acc_button_cmd(0x4)) # Follow Distance button
+    self.safety.safety_tx_hook(self._acc_button_cmd(flw_dist=1))
     self.assertFalse(self.safety.get_controls_allowed())
     self.safety.set_controls_allowed(1)
-    self.safety.safety_tx_hook(self._acc_button_cmd(0x8)) # Set button
+    self.safety.safety_tx_hook(self._acc_button_cmd(_set=1))
     self.assertFalse(self.safety.get_controls_allowed())
     self.safety.set_controls_allowed(1)
-    self.safety.safety_tx_hook(self._acc_button_cmd(0x10)) # Res button
+    self.safety.safety_tx_hook(self._acc_button_cmd(res=1))
     self.assertFalse(self.safety.get_controls_allowed())
     self.safety.set_controls_allowed(1)
-    self.safety.safety_tx_hook(self._acc_button_cmd(0x20)) # No button pressed
+    self.safety.safety_tx_hook(self._acc_button_cmd())
     self.assertFalse(self.safety.get_controls_allowed())
 
-  def test_relay_malfunction(self):
-    StdTest.test_relay_malfunction(self, 0x169)
-
-  def test_fwd_hook(self):
-
-    buss = list(range(0x0, 0x3))
-    msgs = list(range(0x1, 0x800))
-
-    blocked_msgs = [(2, 0x169), (2, 0x2b1), (2, 0x4cc), (0, 0x280)]
-    for b in buss:
-      for m in msgs:
-        if b == 0:
-          fwd_bus = 2
-        elif b == 1:
-          fwd_bus = -1
-        elif b == 2:
-          fwd_bus = 0
-
-        if (b, m) in blocked_msgs:
-          fwd_bus = -1
-
-        # assume len 8
-        self.assertEqual(fwd_bus, self.safety.safety_fwd_hook(b, make_msg(b, m, 8)))
 
 if __name__ == "__main__":
   unittest.main()