CameraITS: Adds result processing to sensor fusion
Output logs are now scraped for time shifts and socket failures. The
mean and standard deviation over all time shifts are calculated, and the
number of socket failures is recorded.
Test: run_sensor_fusion_box.py
Bug: 62838524
Change-Id: Ic8841d4a78a670d561e1ef0e104b3a9231ac4a1f
diff --git a/apps/CameraITS/tools/run_sensor_fusion_box.py b/apps/CameraITS/tools/run_sensor_fusion_box.py
index fbb892e..db48534 100644
--- a/apps/CameraITS/tools/run_sensor_fusion_box.py
+++ b/apps/CameraITS/tools/run_sensor_fusion_box.py
@@ -14,26 +14,35 @@
import os
import os.path
+import re
import subprocess
import sys
import tempfile
import time
import its.device
+import numpy
SCENE_NAME = 'sensor_fusion'
SKIP_RET_CODE = 101
TEST_NAME = 'test_sensor_fusion'
TEST_DIR = os.path.join(os.getcwd(), 'tests', SCENE_NAME)
+W, H = 640, 480
+
+# For finding best correlation shifts from test output logs.
+SHIFT_RE = re.compile('^Best correlation of [0-9.]+ at shift of [-0-9.]+ms$')
+# For finding lines that indicate socket issues in failed test runs.
+SOCKET_FAIL_RE = re.compile(
+ r'.*((socket\.(error|timeout))|(Problem with socket)).*')
FPS = 30
TEST_LENGTH = 7 # seconds
def main():
- """Run all the test_sensor_fusion NUM_RUNS times.
+ """Run test_sensor_fusion NUM_RUNS times.
- Save intermediate files, and producing a summary/report of the results.
+ Save intermediate files and produce a summary/report of the results.
Script should be run from the top-level CameraITS directory.
@@ -52,11 +61,11 @@
camera_id = '0'
fps = str(FPS)
- img_size = '640,480'
+ img_size = '%s,%s' % (W, H)
num_runs = 1
rotator_ids = 'default'
- tmp_dir = None
test_length = str(TEST_LENGTH)
+ tmp_dir = None
for s in sys.argv[1:]:
if s[:7] == 'camera=' and len(s) > 7:
camera_id = s[7:]
@@ -68,10 +77,10 @@
num_runs = int(s[9:])
elif s[:8] == 'rotator=' and len(s) > 8:
rotator_ids = s[8:]
- elif s[:8] == 'tmp_dir=' and len(s) > 8:
- tmp_dir = s[8:]
elif s[:12] == 'test_length=' and len(s) > 12:
test_length = s[12:]
+ elif s[:8] == 'tmp_dir=' and len(s) > 8:
+ tmp_dir = s[8:]
if camera_id not in ['0', '1']:
print 'Need to specify camera 0 or 1'
@@ -91,7 +100,7 @@
print 'Preparing to run sensor_fusion on camera', camera_id
img_size_arg = 'size=' + img_size
- print 'Image dimensions are ' + img_size
+ print 'Image dimensions are ' + 'x'.join(img_size.split(','))
fps_arg = 'fps=' + fps
test_length_arg = 'test_length=' + test_length
@@ -99,12 +108,15 @@
os.mkdir(os.path.join(tmpdir, camera_id))
# Run test "num_runs" times, capturing stdout and stderr.
- numpass = 0
- numfail = 0
- numskip = 0
+ num_pass = 0
+ num_fail = 0
+ num_skip = 0
+ num_socket_fails = 0
+ num_non_socket_fails = 0
+ shift_list = []
for i in range(num_runs):
os.mkdir(os.path.join(tmpdir, camera_id, SCENE_NAME+'_'+str(i)))
- cmd = ('python tools/rotation_rig.py rotator=%s' % rotator_ids)
+ cmd = 'python tools/rotation_rig.py rotator=%s' % rotator_ids
subprocess.Popen(cmd.split())
cmd = ['python', os.path.join(TEST_DIR, TEST_NAME+'.py'),
device_id_arg, camera_id_arg, rotator_id_arg, img_size_arg,
@@ -115,24 +127,96 @@
t0 = time.time()
with open(outpath, 'w') as fout, open(errpath, 'w') as ferr:
retcode = subprocess.call(
- cmd, stderr=ferr, stdout=fout, cwd=outdir)
+ cmd, stderr=ferr, stdout=fout, cwd=outdir)
t1 = time.time()
if retcode == 0:
retstr = 'PASS '
- numpass += 1
+ time_shift = find_time_shift(outpath)
+ shift_list.append(time_shift)
+ num_pass += 1
elif retcode == SKIP_RET_CODE:
retstr = 'SKIP '
- numskip += 1
+ num_skip += 1
else:
retstr = 'FAIL '
- numfail += 1
+ time_shift = find_time_shift(outpath)
+ if time_shift is None:
+ if is_socket_fail(errpath):
+ num_socket_fails += 1
+ else:
+ num_non_socket_fails += 1
+ else:
+ shift_list.append(time_shift)
+ num_fail += 1
msg = '%s %s/%s [%.1fs]' % (retstr, SCENE_NAME, TEST_NAME, t1-t0)
print msg
- test_result = '%d / %d tests passed (%.1f%%)' % (
- numpass+numskip, num_runs, 100*float(numpass+numskip)/num_runs)
- print test_result
+ if num_pass == 1:
+ print 'Best shift is %sms' % shift_list[0]
+ elif num_pass > 1:
+ shift_arr = numpy.array(shift_list)
+ mean, std = numpy.mean(shift_arr), numpy.std(shift_arr)
+ print 'Best shift mean is %sms with std. dev. of %sms' % (mean, std)
+
+ pass_percentage = 100*float(num_pass+num_skip)/num_runs
+ print '%d / %d tests passed (%.1f%%)' % (num_pass+num_skip,
+ num_runs,
+ pass_percentage)
+
+ if num_socket_fails != 0:
+ print '%s failure(s) due to socket issues' % num_socket_fails
+ if num_non_socket_fails != 0:
+ print '%s non-socket failure(s)' % num_non_socket_fails
+
+
+def is_socket_fail(err_file_path):
+ """Search through a test run's stderr log for any mention of socket issues.
+
+ Args:
+ err_file_path: File path for stderr logs to search through
+
+ Returns:
+ True if the test run failed and it was due to socket issues. Otherwise,
+ False.
+ """
+ return find_matching_line(err_file_path, SOCKET_FAIL_RE) is not None
+
+
+def find_time_shift(out_file_path):
+ """Search through a test run's stdout log for the best time shift.
+
+ Args:
+ out_file_path: File path for stdout logs to search through
+
+ Returns:
+ The best time shift, if one is found. Otherwise, returns None.
+ """
+ line = find_matching_line(out_file_path, SHIFT_RE)
+ if line is None:
+ return None
+ else:
+ words = line.split(' ')
+ # Get last word and strip off 'ms\n' before converting to a float.
+ return float(words[-1][:-3])
+
+
+def find_matching_line(file_path, regex):
+ """Search each line in the file at 'file_path' for a line matching 'regex'.
+
+ Args:
+ file_path: File path for file being searched
+ regex: Regex used to match against lines
+
+ Returns:
+ The first matching line. If none exists, returns None.
+ """
+ with open(file_path) as f:
+ for line in f:
+ if regex.match(line):
+ return line
+ return None
+
if __name__ == '__main__':
main()
