panda safety test that replays drives of saved CAN messages (commaai#151

)

* panda safety test that replays drives of saved CAN messages

* utility to trim Cabana CSV logs to just messages relevant for panda safety testing

* when trimming, only output the same line once even if it matches both criteria

tests/safety/test_chrysler.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python2
+import csv
+import glob
 import unittest
 import numpy as np
 import libpandasafety_py
@@ -24,6 +26,11 @@ def sign(a):
   else:
     return -1
 
+def swap_bytes(data_str):
+  """Accepts string with hex, returns integer with order swapped for CAN."""
+  a = int(data_str, 16)
+  return ((a & 0xff) << 24) + ((a & 0xff00) << 8) + ((a & 0x00ff0000) >> 8) + ((a & 0xff000000) >> 24)
+
 class TestChryslerSafety(unittest.TestCase):
   @classmethod
   def setUp(cls):
@@ -166,6 +173,33 @@ def test_torque_measurements(self):
     self.assertEqual(0, self.safety.get_chrysler_torque_meas_max())
     self.assertEqual(0, self.safety.get_chrysler_torque_meas_min())
 
+  def _replay_drive(self, csv_reader):
+    for row in csv_reader:
+      if len(row) != 4:  # sometimes truncated at end of the file
+        continue
+      if row[0] == 'time':  # skip CSV header
+        continue
+      addr = int(row[1])
+      bus = int(row[2])
+      data_str = row[3]  # Example '081407ff0806e06f'
+      to_send = libpandasafety_py.ffi.new('CAN_FIFOMailBox_TypeDef *')
+      to_send[0].RIR = addr << 21
+      to_send[0].RDHR = swap_bytes(data_str[8:])
+      to_send[0].RDLR = swap_bytes(data_str[:8])
+      if (bus == 128):
+        self.assertTrue(self.safety.chrysler_tx_hook(to_send), msg=row)
+      else:
+        self.safety.chrysler_rx_hook(to_send)
+
+  def test_replay_drive(self):
+    # In Cabana, click "Save Log" and then put the downloaded CSV in this directory.
+    test_files = glob.glob('chrysler_*.csv')
+    for filename in test_files:
+      print 'testing %s' % filename
+      with open(filename) as csvfile:
+        reader = csv.reader(csvfile)
+        self._replay_drive(reader)
+
 
 if __name__ == "__main__":
   unittest.main()

tests/safety/trim_csv.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python2
+
+# This trims CAN message CSV files to just the messages relevant for Panda testing.
+# Usage:
+# cat input.csv | ./trim_csv.py > output.csv
+import fileinput
+
+addr_to_keep = [544, 0x1f4, 0x292]  # For Chrysler, update to the addresses that matter for you.
+
+for line in fileinput.input():
+  line = line.strip()
+  cols = line.split(',')
+  if len(cols) != 4:
+    continue  # malformed, such as at the end or every 60s.
+  (_, addr, bus, _) = cols
+  if (addr == 'addr'):
+    continue
+  if (int(bus) == 128):  # Keep all messages sent by OpenPilot.
+    print line
+  elif (int(addr) in addr_to_keep):
+    print line