# Copyright 2016 The Android Open Source Project
|
#
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
# you may not use this file except in compliance with the License.
|
# You may obtain a copy of the License at
|
#
|
# http://www.apache.org/licenses/LICENSE-2.0
|
#
|
# Unless required by applicable law or agreed to in writing, software
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
# See the License for the specific language governing permissions and
|
# limitations under the License.
|
|
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 test_sensor_fusion NUM_RUNS times.
|
|
Save intermediate files and produce a summary/report of the results.
|
|
Script should be run from the top-level CameraITS directory.
|
|
Command line arguments:
|
camera: Camera(s) to be tested. Use comma to separate multiple
|
camera Ids. Ex: 'camera=0,1' or 'camera=1'
|
device: Device id for adb
|
fps: FPS to capture with during the test
|
img_size: Comma-separated dimensions of captured images (defaults to
|
640x480). Ex: 'img_size=<width>,<height>'
|
num_runs: Number of times to repeat the test
|
rotator: String for rotator id in for vid:pid:ch
|
test_length: How long the test should run for (in seconds)
|
tmp_dir: Location of temp directory for output files
|
"""
|
|
camera_id = '0'
|
fps = str(FPS)
|
img_size = '%s,%s' % (W, H)
|
num_runs = 1
|
rotator_ids = 'default'
|
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:]
|
if s[:4] == 'fps=' and len(s) > 4:
|
fps = s[4:]
|
elif s[:9] == 'img_size=' and len(s) > 9:
|
img_size = s[9:]
|
elif s[:9] == 'num_runs=' and len(s) > 9:
|
num_runs = int(s[9:])
|
elif s[:8] == 'rotator=' and len(s) > 8:
|
rotator_ids = 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:]
|
|
# Make output directories to hold the generated files.
|
tmpdir = tempfile.mkdtemp(dir=tmp_dir)
|
print 'Saving output files to:', tmpdir, '\n'
|
|
device_id = its.device.get_device_id()
|
device_id_arg = 'device=' + device_id
|
print 'Testing device ' + device_id
|
|
# ensure camera_id is valid
|
avail_camera_ids = find_avail_camera_ids()
|
if camera_id not in avail_camera_ids:
|
print 'Need to specify valid camera_id in ', avail_camera_ids
|
sys.exit()
|
|
camera_id_arg = 'camera=' + camera_id
|
if rotator_ids:
|
rotator_id_arg = 'rotator=' + rotator_ids
|
print 'Preparing to run sensor_fusion on camera', camera_id
|
|
img_size_arg = 'img_size=' + img_size
|
print 'Image dimensions are ' + 'x'.join(img_size.split(','))
|
|
fps_arg = 'fps=' + fps
|
test_length_arg = 'test_length=' + test_length
|
print 'Capturing at %sfps' % fps
|
|
os.mkdir(os.path.join(tmpdir, camera_id))
|
|
# Run test "num_runs" times, capturing stdout and stderr.
|
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
|
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,
|
fps_arg, test_length_arg]
|
outdir = os.path.join(tmpdir, camera_id, SCENE_NAME+'_'+str(i))
|
outpath = os.path.join(outdir, TEST_NAME+'_stdout.txt')
|
errpath = os.path.join(outdir, TEST_NAME+'_stderr.txt')
|
t0 = time.time()
|
with open(outpath, 'w') as fout, open(errpath, 'w') as ferr:
|
retcode = subprocess.call(
|
cmd, stderr=ferr, stdout=fout, cwd=outdir)
|
t1 = time.time()
|
|
if retcode == 0:
|
retstr = 'PASS '
|
time_shift = find_time_shift(outpath)
|
shift_list.append(time_shift)
|
num_pass += 1
|
elif retcode == SKIP_RET_CODE:
|
retstr = 'SKIP '
|
num_skip += 1
|
else:
|
retstr = 'FAIL '
|
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
|
|
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
|
|
def find_avail_camera_ids():
|
"""Find the available camera IDs.
|
|
Returns:
|
list of available cameras
|
"""
|
with its.device.ItsSession() as cam:
|
avail_camera_ids = cam.get_camera_ids()
|
return avail_camera_ids
|
|
|
if __name__ == '__main__':
|
main()
|
