camerad jenkins box test (#20413)

* separate test

* send

* update test

* snapshot is rgb

* this does not belong here

* clean up

* tici

* no if

* no black frames

* flaky leds

* rename and cleanup

* should be stable and avoid lucky pass

* print

* logical and

* run in jenkins

* source profile on tici

Co-authored-by: Comma Device <device@comma.ai>
Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>
old-commit-hash: 97133a8fac

Jenkinsfile

@@ -11,6 +11,9 @@ export GIT_BRANCH=${env.GIT_BRANCH}
 export GIT_COMMIT=${env.GIT_COMMIT}
 
 source ~/.bash_profile
+if [ -f /TICI ]; then
+  source /etc/profile
+fi
 
 ln -snf ${env.TEST_DIR} /data/pythonpath
 
@@ -151,7 +154,6 @@ pipeline {
                       ["test boardd loopback", "nosetests -s selfdrive/boardd/tests/test_boardd_loopback.py"],
                       ["test loggerd", "python selfdrive/loggerd/tests/test_loggerd.py"],
                       ["test encoder", "python selfdrive/loggerd/tests/test_encoder.py"],
-                      ["test camerad", "python selfdrive/camerad/test/test_camerad.py"],
                       ["test logcatd", "python selfdrive/logcatd/tests/test_logcatd_android.py"],
                       //["test updater", "python installer/updater/test_updater.py"],
                     ])
@@ -167,12 +169,31 @@ pipeline {
                       ["build", "SCONS_CACHE=1 scons -j16"],
                       ["test loggerd", "python selfdrive/loggerd/tests/test_loggerd.py"],
                       ["test encoder", "LD_LIBRARY_PATH=/usr/local/lib python selfdrive/loggerd/tests/test_encoder.py"],
-                      ["test camerad", "python selfdrive/camerad/test/test_camerad.py"],
                       //["build release3-staging", "cd release && PUSH=${env.R3_PUSH} ./build_release3.sh"],
                     ])
                   }
                 }
 
+                stage('camerad') {
+                  steps {
+                    phone_steps("eon-party", [
+                      ["build", "SCONS_CACHE=1 scons -j16"],
+                      ["test camerad", "python selfdrive/camerad/test/test_camerad.py"],
+                      ["test exposure", "python selfdrive/camerad/test/test_exposure.py"],
+                    ])
+                  }
+                }
+
+                stage('Tici camerad') {
+                  steps {
+                    phone_steps("tici-party", [
+                      ["build", "SCONS_CACHE=1 scons -j16"],
+                      ["test camerad", "python selfdrive/camerad/test/test_camerad.py"],
+                      ["test exposure", "python selfdrive/camerad/test/test_exposure.py"],
+                    ])
+                  }
+                }
+
               }
             }
           }

selfdrive/camerad/test/test_camerad.py

@@ -2,11 +2,9 @@
 
 import time
 import unittest
-import numpy as np
 
 import cereal.messaging as messaging
 from selfdrive.test.helpers import with_processes
-from selfdrive.camerad.snapshot.snapshot import get_snapshots
 
 # only tests for EON and TICI
 from selfdrive.hardware import EON, TICI
@@ -31,70 +29,6 @@ class TestCamerad(unittest.TestCase):
     if not (EON or TICI):
       raise unittest.SkipTest
 
-    # assert "SEND_REAR" in os.environ
-    # assert "SEND_FRONT" in os.environ
-
-  def _numpy_bgr2gray(self, im):
-    ret = np.clip(im[:,:,0] * 0.114 + im[:,:,1] * 0.587 + im[:,:,2] * 0.299, 0, 255).astype(np.uint8)
-    return ret
-
-  def _numpy_lap(self, im):
-    ret = np.zeros(im.shape)
-    ret += -4 * im
-    ret += np.concatenate([np.zeros((im.shape[0],1)),im[:,:-1]], axis=1)
-    ret += np.concatenate([im[:,1:],np.zeros((im.shape[0],1))], axis=1)
-    ret += np.concatenate([np.zeros((1,im.shape[1])),im[:-1,:]], axis=0)
-    ret += np.concatenate([im[1:,:],np.zeros((1,im.shape[1]))], axis=0)
-    ret = np.clip(ret, 0, 255).astype(np.uint8)
-    return ret
-
-  def _is_really_sharp(self, i, threshold=800, roi_max=np.array([8,6]), roi_xxyy=np.array([1,6,2,3])):
-    i = self._numpy_bgr2gray(i)
-    x_pitch = i.shape[1] // roi_max[0]
-    y_pitch = i.shape[0] // roi_max[1]
-    lap = self._numpy_lap(i)
-    lap_map = np.zeros((roi_max[1], roi_max[0]))
-    for r in range(lap_map.shape[0]):
-      for c in range(lap_map.shape[1]):
-        selected_lap = lap[r*y_pitch:(r+1)*y_pitch, c*x_pitch:(c+1)*x_pitch]
-        lap_map[r][c] = 5*selected_lap.var() + selected_lap.max()
-    print(lap_map[roi_xxyy[2]:roi_xxyy[3]+1,roi_xxyy[0]:roi_xxyy[1]+1])
-    if (lap_map[roi_xxyy[2]:roi_xxyy[3]+1,roi_xxyy[0]:roi_xxyy[1]+1] > threshold).sum() > \
-          (roi_xxyy[1]+1-roi_xxyy[0]) * (roi_xxyy[3]+1-roi_xxyy[2]) * 0.9:
-      return True
-    else:
-      return False
-
-  def _is_exposure_okay(self, i, med_ex=np.array([0.2,0.4]), mean_ex=np.array([0.2,0.6])):
-    i = self._numpy_bgr2gray(i)
-    i_median = np.median(i) / 256
-    i_mean = np.mean(i) / 256
-    print([i_median, i_mean])
-    return med_ex[0] < i_median < med_ex[1] and mean_ex[0] < i_mean < mean_ex[1]
-
-  @unittest.skip # skip for now
-  @with_processes(['camerad'])
-  def test_camera_operation(self):
-    print("checking image outputs")
-    if EON:
-      # run checks similar to prov
-      time.sleep(15) # wait for startup and AF
-      pic, fpic = get_snapshots()
-      self.assertTrue(self._is_really_sharp(pic))
-      self.assertTrue(self._is_exposure_okay(pic))
-      self.assertTrue(self._is_exposure_okay(fpic))
-
-      time.sleep(30)
-      # check again for consistency
-      pic, fpic = get_snapshots()
-      self.assertTrue(self._is_really_sharp(pic))
-      self.assertTrue(self._is_exposure_okay(pic))
-      self.assertTrue(self._is_exposure_okay(fpic))
-    elif TICI:
-      raise unittest.SkipTest # TBD
-    else:
-      raise unittest.SkipTest
-
   @with_processes(['camerad'])
   def test_frame_packets(self):
     print("checking frame pkts continuity")

selfdrive/camerad/test/test_exposure.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+
+import time
+import unittest
+import os
+import numpy as np
+
+from selfdrive.test.helpers import with_processes
+from selfdrive.camerad.snapshot.snapshot import get_snapshots
+
+from selfdrive.hardware import EON, TICI
+
+TEST_TIME = 45
+REPEAT = 5
+
+os.environ["SEND_ROAD"] = "1"
+os.environ["SEND_DRIVER"] = "1"
+if TICI:
+  os.environ["SEND_WIDE_ROAD"] = "1"
+
+class TestCamerad(unittest.TestCase):
+  @classmethod
+  def setUpClass(cls):
+    if not (EON or TICI):
+      raise unittest.SkipTest
+
+  def _numpy_rgb2gray(self, im):
+    ret = np.clip(im[:,:,2] * 0.114 + im[:,:,1] * 0.587 + im[:,:,0] * 0.299, 0, 255).astype(np.uint8)
+    return ret
+
+  def _is_exposure_okay(self, i, med_mean=np.array([[0.2,0.4],[0.2,0.6]])):
+    h, w = i.shape[:2]
+    i = i[h//10:9*h//10,w//10:9*w//10]
+    med_ex, mean_ex = med_mean
+    i = self._numpy_rgb2gray(i)
+    i_median = np.median(i) / 255.
+    i_mean = np.mean(i) / 255.
+    print([i_median, i_mean])
+    return med_ex[0] < i_median < med_ex[1] and mean_ex[0] < i_mean < mean_ex[1]
+
+
+  @with_processes(['camerad'])
+  def test_camera_operation(self):
+    print("checking image outputs")
+
+    start = time.time()
+    passed = 0
+    while(time.time() - start < TEST_TIME and passed < REPEAT):
+      rpic, dpic = get_snapshots(frame="roadCameraState", front_frame="driverCameraState")
+
+      res = self._is_exposure_okay(rpic)
+      res = res and self._is_exposure_okay(dpic)
+
+      if TICI:
+        wpic, _ = get_snapshots(frame="wideRoadCameraState")
+        res = res and self._is_exposure_okay(wpic)
+
+      if passed > 0 and not res:
+        passed = -passed # fails test if any failure after first sus
+        break
+
+      passed += int(res)
+      time.sleep(2)
+    print(passed)
+    self.assertTrue(passed >= REPEAT)
+
+if __name__ == "__main__":
+  unittest.main()