# Copyright 2014 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 copy
import math
import os
import os.path
import re
import subprocess
import sys
import tempfile
import threading
import time
import its.caps
import its.cv2image
import its.device
from its.device import ItsSession
import its.image
import numpy as np
# For sanity checking the installed APK's target SDK version
MIN_SUPPORTED_SDK_VERSION = 28 # P
CHART_DELAY = 1 # seconds
CHART_DISTANCE = 30.0 # cm
CHART_HEIGHT = 13.5 # cm
CHART_LEVEL = 96
CHART_SCALE_START = 0.65
CHART_SCALE_STOP = 1.35
CHART_SCALE_STEP = 0.025
FACING_EXTERNAL = 2
NUM_TRYS = 2
PROC_TIMEOUT_CODE = -101 # terminated process return -process_id
PROC_TIMEOUT_TIME = 900 # timeout in seconds for a process (15 minutes)
SCENE3_FILE = os.path.join(os.environ['CAMERA_ITS_TOP'], 'pymodules', 'its',
'test_images', 'ISO12233.png')
SKIP_RET_CODE = 101 # note this must be same as tests/scene*/test_*
VGA_HEIGHT = 480
VGA_WIDTH = 640
# Not yet mandated tests
NOT_YET_MANDATED = {
'scene0': [
'test_test_patterns',
'test_tonemap_curve'
],
'scene1': [
'test_ae_precapture_trigger',
'test_channel_saturation'
],
'scene2': [
'test_auto_per_frame_control'
],
'scene2b': [],
'scene2c': [],
'scene3': [],
'scene4': [],
'scene5': [],
'sensor_fusion': []
}
# Must match mHiddenPhysicalCameraSceneIds in ItsTestActivity.java
HIDDEN_PHYSICAL_CAMERA_TESTS = {
'scene0': [
'test_burst_capture',
'test_metadata',
'test_read_write',
'test_sensor_events'
],
'scene1': [
'test_exposure',
'test_dng_noise_model',
'test_linearity',
'test_raw_exposure',
'test_raw_sensitivity'
],
'scene2': [
'test_faces',
'test_num_faces'
],
'scene2b': [],
'scene2c': [],
'scene3': [],
'scene4': [
'test_aspect_ratio_and_crop'
],
'scene5': [],
'sensor_fusion': [
'test_sensor_fusion'
]
}
def run_subprocess_with_timeout(cmd, fout, ferr, outdir):
"""Run subprocess with a timeout.
Args:
cmd: list containing python command
fout: stdout file for the test
ferr: stderr file for the test
outdir: dir location for fout/ferr
Returns:
process status or PROC_TIMEOUT_CODE if timer maxes
"""
proc = subprocess.Popen(
cmd, stdout=fout, stderr=ferr, cwd=outdir)
timer = threading.Timer(PROC_TIMEOUT_TIME, proc.kill)
try:
timer.start()
proc.communicate()
test_code = proc.returncode
finally:
timer.cancel()
if test_code < 0:
return PROC_TIMEOUT_CODE
else:
return test_code
def calc_camera_fov(camera_id, hidden_physical_id):
"""Determine the camera field of view from internal params."""
with ItsSession(camera_id, hidden_physical_id) as cam:
props = cam.get_camera_properties()
props = cam.override_with_hidden_physical_camera_props(props)
focal_ls = props['android.lens.info.availableFocalLengths']
if len(focal_ls) > 1:
print 'Doing capture to determine logical camera focal length'
cap = cam.do_capture(its.objects.auto_capture_request())
focal_l = cap['metadata']['android.lens.focalLength']
else:
focal_l = focal_ls[0]
sensor_size = props['android.sensor.info.physicalSize']
diag = math.sqrt(sensor_size['height'] ** 2 +
sensor_size['width'] ** 2)
try:
fov = str(round(2 * math.degrees(math.atan(diag / (2 * focal_l))), 2))
except ValueError:
fov = str(0)
print 'Calculated FoV: %s' % fov
return fov
def evaluate_socket_failure(err_file_path):
"""Determine if test fails due to socket FAIL."""
socket_fail = False
with open(err_file_path, 'r') as ferr:
for line in ferr:
if (line.find('socket.error') != -1 or
line.find('socket.timeout') != -1 or
line.find('Problem with socket') != -1):
socket_fail = True
return socket_fail
def skip_sensor_fusion(camera_id):
"""Determine if sensor fusion test is skipped for this camera."""
skip_code = SKIP_RET_CODE
with ItsSession(camera_id) as cam:
props = cam.get_camera_properties()
if (its.caps.sensor_fusion(props) and its.caps.manual_sensor(props) and
props['android.lens.facing'] is not FACING_EXTERNAL):
skip_code = None
return skip_code
def main():
"""Run all the automated tests, saving intermediate files, and producing
a summary/report of the results.
Script should be run from the top-level CameraITS directory.
Command line arguments:
camera: the camera(s) to be tested. Use comma to separate multiple
camera Ids. Ex: "camera=0,1" or "camera=1"
device: device id for adb
scenes: the test scene(s) to be executed. Use comma to separate
multiple scenes. Ex: "scenes=scene0,scene1" or
"scenes=0,1,sensor_fusion" (sceneX can be abbreviated by X
where X is a integer)
chart: [Experimental] another android device served as test chart
display. When this argument presents, change of test scene
will be handled automatically. Note that this argument
requires special physical/hardware setup to work and may not
work on all android devices.
result: Device ID to forward results to (in addition to the device
that the tests are running on).
rot_rig: [Experimental] ID of the rotation rig being used (formatted as
"<vendor ID>:<product ID>:<channel #>" or "default")
tmp_dir: location of temp directory for output files
skip_scene_validation: force skip scene validation. Used when test scene
is setup up front and don't require tester validation.
dist: [Experimental] chart distance in cm.
"""
all_scenes = ["scene0", "scene1", "scene2", "scene2b", "scene2c", "scene3", "scene4", "scene5",
"sensor_fusion"]
auto_scenes = ["scene0", "scene1", "scene2", "scene2b", "scene2c", "scene3", "scene4"]
scene_req = {
"scene0": None,
"scene1": "A grey card covering at least the middle 30% of the scene",
"scene2": "A picture containing human faces",
"scene2b": "A picture containing human faces",
"scene2c": "A picture containing human faces",
"scene3": "The ISO 12233 chart",
"scene4": "A specific test page of a circle covering at least the "
"middle 50% of the scene. See CameraITS.pdf section 2.3.4 "
"for more details",
"scene5": "Capture images with a diffuser attached to the camera. See "
"CameraITS.pdf section 2.3.4 for more details",
"sensor_fusion": "Rotating checkboard pattern. See "
"sensor_fusion/SensorFusion.pdf for detailed "
"instructions.\nNote that this test will be skipped "
"on devices not supporting REALTIME camera timestamp."
}
scene_extra_args = {
"scene5": ["doAF=False"]
}
camera_id_combos = []
scenes = []
chart_host_id = None
result_device_id = None
rot_rig_id = None
tmp_dir = None
skip_scene_validation = False
chart_distance = CHART_DISTANCE
chart_level = CHART_LEVEL
one_camera_argv = sys.argv[1:]
for s in list(sys.argv[1:]):
if s[:7] == "camera=" and len(s) > 7:
camera_ids = s[7:].split(',')
camera_id_combos = its.device.parse_camera_ids(camera_ids)
one_camera_argv.remove(s)
elif s[:7] == "scenes=" and len(s) > 7:
scenes = s[7:].split(',')
elif s[:6] == 'chart=' and len(s) > 6:
chart_host_id = s[6:]
elif s[:7] == 'result=' and len(s) > 7:
result_device_id = s[7:]
elif s[:8] == 'rot_rig=' and len(s) > 8:
rot_rig_id = s[8:] # valid values: 'default' or '$VID:$PID:$CH'
# The default '$VID:$PID:$CH' is '04d8:fc73:1'
elif s[:8] == 'tmp_dir=' and len(s) > 8:
tmp_dir = s[8:]
elif s == 'skip_scene_validation':
skip_scene_validation = True
elif s[:5] == 'dist=' and len(s) > 5:
chart_distance = float(re.sub('cm', '', s[5:]))
elif s[:11] == 'brightness=' and len(s) > 11:
chart_level = s[11:]
chart_dist_arg = 'dist= ' + str(chart_distance)
chart_level_arg = 'brightness=' + str(chart_level)
auto_scene_switch = chart_host_id is not None
merge_result_switch = result_device_id is not None
# Run through all scenes if user does not supply one
possible_scenes = auto_scenes if auto_scene_switch else all_scenes
if not scenes:
scenes = possible_scenes
else:
# Validate user input scene names
valid_scenes = True
temp_scenes = []
for s in scenes:
if s in possible_scenes:
temp_scenes.append(s)
else:
try:
# Try replace "X" to "sceneX"
scene_str = "scene" + s
if scene_str not in possible_scenes:
valid_scenes = False
break
temp_scenes.append(scene_str)
except ValueError:
valid_scenes = False
break
if not valid_scenes:
print 'Unknown scene specified:', s
assert False
scenes = temp_scenes
# Initialize test results
results = {}
result_key = ItsSession.RESULT_KEY
for s in all_scenes:
results[s] = {result_key: ItsSession.RESULT_NOT_EXECUTED}
# Make output directories to hold the generated files.
topdir = tempfile.mkdtemp(dir=tmp_dir)
subprocess.call(['chmod', 'g+rx', topdir])
print "Saving output files to:", topdir, "\n"
device_id = its.device.get_device_id()
device_id_arg = "device=" + device_id
print "Testing device " + device_id
# Sanity check CtsVerifier SDK level
# Here we only do warning as there is no guarantee on pm dump output formt not changed
# Also sometimes it's intentional to run mismatched versions
cmd = "adb -s %s shell pm dump com.android.cts.verifier" % (device_id)
dump_path = os.path.join(topdir, 'CtsVerifier.txt')
with open(dump_path, 'w') as fout:
fout.write('ITS minimum supported SDK version is %d\n--\n' % (MIN_SUPPORTED_SDK_VERSION))
fout.flush()
ret_code = subprocess.call(cmd.split(), stdout=fout)
if ret_code != 0:
print "Warning: cannot get CtsVerifier SDK version. Is CtsVerifier installed?"
ctsv_version = None
ctsv_version_name = None
with open(dump_path, 'r') as f:
target_sdk_found = False
version_name_found = False
for line in f:
match = re.search('targetSdk=([0-9]+)', line)
if match:
ctsv_version = int(match.group(1))
target_sdk_found = True
match = re.search('versionName=([\S]+)$', line)
if match:
ctsv_version_name = match.group(1)
version_name_found = True
if target_sdk_found and version_name_found:
break
if ctsv_version is None:
print "Warning: cannot get CtsVerifier SDK version. Is CtsVerifier installed?"
elif ctsv_version < MIN_SUPPORTED_SDK_VERSION:
print "Warning: CtsVerifier version (%d) < ITS version (%d), is this intentional?" % (
ctsv_version, MIN_SUPPORTED_SDK_VERSION)
else:
print "CtsVerifier targetSdk is", ctsv_version
if ctsv_version_name:
print "CtsVerifier version name is", ctsv_version_name
# Hard check on ItsService/host script version that should catch incompatible APK/script
with ItsSession() as cam:
cam.check_its_version_compatible()
# Sanity Check for devices
device_bfp = its.device.get_device_fingerprint(device_id)
assert device_bfp is not None
if auto_scene_switch:
chart_host_bfp = its.device.get_device_fingerprint(chart_host_id)
assert chart_host_bfp is not None
if merge_result_switch:
result_device_bfp = its.device.get_device_fingerprint(result_device_id)
assert_err_msg = ('Cannot merge result to a different build, from '
'%s to %s' % (device_bfp, result_device_bfp))
assert device_bfp == result_device_bfp, assert_err_msg
# user doesn't specify camera id, run through all cameras
if not camera_id_combos:
with its.device.ItsSession() as cam:
camera_ids = cam.get_camera_ids()
camera_id_combos = its.device.parse_camera_ids(camera_ids);
print "Running ITS on camera: %s, scene %s" % (camera_id_combos, scenes)
if auto_scene_switch:
# merge_result only supports run_parallel_tests
if merge_result_switch and camera_ids[0] == "1":
print "Skip chart screen"
time.sleep(1)
else:
print "Waking up chart screen: ", chart_host_id
screen_id_arg = ("screen=%s" % chart_host_id)
cmd = ["python", os.path.join(os.environ["CAMERA_ITS_TOP"], "tools",
"wake_up_screen.py"), screen_id_arg,
chart_level_arg]
wake_code = subprocess.call(cmd)
assert wake_code == 0
for id_combo in camera_id_combos:
camera_fov = calc_camera_fov(id_combo.id, id_combo.sub_id)
id_combo_string = id_combo.id;
has_hidden_sub_camera = id_combo.sub_id is not None
if has_hidden_sub_camera:
id_combo_string += ":" + id_combo.sub_id
scenes = [scene for scene in scenes if HIDDEN_PHYSICAL_CAMERA_TESTS[scene]]
# Loop capturing images until user confirm test scene is correct
camera_id_arg = "camera=" + id_combo.id
print "Preparing to run ITS on camera", id_combo_string, "for scenes ", scenes
os.mkdir(os.path.join(topdir, id_combo_string))
for d in scenes:
os.mkdir(os.path.join(topdir, id_combo_string, d))
tot_tests = []
tot_pass = 0
for scene in scenes:
skip_code = None
tests = [(s[:-3], os.path.join("tests", scene, s))
for s in os.listdir(os.path.join("tests", scene))
if s[-3:] == ".py" and s[:4] == "test"]
tests.sort()
tot_tests.extend(tests)
summary = "Cam" + id_combo_string + " " + scene + "\n"
numpass = 0
numskip = 0
num_not_mandated_fail = 0
numfail = 0
validate_switch = True
if scene_req[scene] is not None:
out_path = os.path.join(topdir, id_combo_string, scene+".jpg")
out_arg = "out=" + out_path
if scene == 'sensor_fusion':
skip_code = skip_sensor_fusion(id_combo.id)
if rot_rig_id or skip_code == SKIP_RET_CODE:
validate_switch = False
if skip_scene_validation:
validate_switch = False
cmd = None
if auto_scene_switch:
if (not merge_result_switch or
(merge_result_switch and id_combo_string == '0')):
scene_arg = 'scene=' + scene
fov_arg = 'fov=' + camera_fov
cmd = ['python',
os.path.join(os.getcwd(), 'tools/load_scene.py'),
scene_arg, chart_dist_arg, fov_arg, screen_id_arg]
else:
time.sleep(CHART_DELAY)
else:
# Skip scene validation under certain conditions
if validate_switch and not merge_result_switch:
scene_arg = 'scene=' + scene_req[scene]
extra_args = scene_extra_args.get(scene, [])
cmd = ['python',
os.path.join(os.getcwd(),
'tools/validate_scene.py'),
camera_id_arg, out_arg,
scene_arg, device_id_arg] + extra_args
if cmd is not None:
valid_scene_code = subprocess.call(cmd, cwd=topdir)
assert valid_scene_code == 0
print 'Start running ITS on camera %s, %s' % (id_combo_string, scene)
# Extract chart from scene for scene3 once up front
chart_loc_arg = ''
chart_height = CHART_HEIGHT
if scene == 'scene3':
chart_height *= its.cv2image.calc_chart_scaling(
chart_distance, camera_fov)
chart = its.cv2image.Chart(SCENE3_FILE, chart_height,
chart_distance, CHART_SCALE_START,
CHART_SCALE_STOP, CHART_SCALE_STEP,
id_combo.id)
chart_loc_arg = 'chart_loc=%.2f,%.2f,%.2f,%.2f,%.3f' % (
chart.xnorm, chart.ynorm, chart.wnorm, chart.hnorm,
chart.scale)
# Run each test, capturing stdout and stderr.
for (testname, testpath) in tests:
# Only pick predefined tests for hidden physical camera
if has_hidden_sub_camera and \
testname not in HIDDEN_PHYSICAL_CAMERA_TESTS[scene]:
numskip += 1
continue
if auto_scene_switch:
if merge_result_switch and id_combo_string == '0':
# Send an input event to keep the screen not dimmed.
# Since we are not using camera of chart screen, FOCUS event
# should do nothing but keep the screen from dimming.
# The "sleep after x minutes of inactivity" display setting
# determines how long this command can keep screen bright.
# Setting it to something like 30 minutes should be enough.
cmd = ('adb -s %s shell input keyevent FOCUS'
% chart_host_id)
subprocess.call(cmd.split())
t0 = time.time()
for num_try in range(NUM_TRYS):
outdir = os.path.join(topdir, id_combo_string, scene)
outpath = os.path.join(outdir, testname+'_stdout.txt')
errpath = os.path.join(outdir, testname+'_stderr.txt')
if scene == 'sensor_fusion':
if skip_code is not SKIP_RET_CODE:
if rot_rig_id:
print 'Rotating phone w/ rig %s' % rot_rig_id
rig = ('python tools/rotation_rig.py rotator=%s' %
rot_rig_id)
subprocess.Popen(rig.split())
else:
print 'Rotate phone 15s as shown in SensorFusion.pdf'
else:
test_code = skip_code
if skip_code is not SKIP_RET_CODE:
cmd = ['python', os.path.join(os.getcwd(), testpath)]
cmd += one_camera_argv + ["camera="+id_combo_string] + [chart_loc_arg]
cmd += [chart_dist_arg]
with open(outpath, 'w') as fout, open(errpath, 'w') as ferr:
test_code = run_subprocess_with_timeout(
cmd, fout, ferr, outdir)
if test_code == 0 or test_code == SKIP_RET_CODE:
break
else:
socket_fail = evaluate_socket_failure(errpath)
if socket_fail or test_code == PROC_TIMEOUT_CODE:
if num_try != NUM_TRYS-1:
print ' Retry %s/%s' % (scene, testname)
else:
break
else:
break
t1 = time.time()
test_failed = False
if test_code == 0:
retstr = "PASS "
numpass += 1
elif test_code == SKIP_RET_CODE:
retstr = "SKIP "
numskip += 1
elif test_code != 0 and testname in NOT_YET_MANDATED[scene]:
retstr = "FAIL*"
num_not_mandated_fail += 1
else:
retstr = "FAIL "
numfail += 1
test_failed = True
msg = "%s %s/%s [%.1fs]" % (retstr, scene, testname, t1-t0)
print msg
its.device.adb_log(device_id, msg)
msg_short = "%s %s [%.1fs]" % (retstr, testname, t1-t0)
if test_failed:
summary += msg_short + "\n"
if numskip > 0:
skipstr = ", %d test%s skipped" % (
numskip, "s" if numskip > 1 else "")
else:
skipstr = ""
test_result = "\n%d / %d tests passed (%.1f%%)%s" % (
numpass + num_not_mandated_fail, len(tests) - numskip,
100.0 * float(numpass + num_not_mandated_fail) /
(len(tests) - numskip)
if len(tests) != numskip else 100.0, skipstr)
print test_result
if num_not_mandated_fail > 0:
msg = "(*) tests are not yet mandated"
print msg
tot_pass += numpass
print "%s compatibility score: %.f/100\n" % (
scene, 100.0 * numpass / len(tests))
summary_path = os.path.join(topdir, id_combo_string, scene, "summary.txt")
with open(summary_path, "w") as f:
f.write(summary)
passed = numfail == 0
results[scene][result_key] = (ItsSession.RESULT_PASS if passed
else ItsSession.RESULT_FAIL)
results[scene][ItsSession.SUMMARY_KEY] = summary_path
if tot_tests:
print "Compatibility Score: %.f/100" % (100.0 * tot_pass / len(tot_tests))
else:
print "Compatibility Score: 0/100"
msg = "Reporting ITS result to CtsVerifier"
print msg
its.device.adb_log(device_id, msg)
if merge_result_switch:
# results are modified by report_result
results_backup = copy.deepcopy(results)
its.device.report_result(result_device_id, id_combo_string, results_backup)
# Report hidden_physical_id results as well.
its.device.report_result(device_id, id_combo_string, results)
if auto_scene_switch:
if merge_result_switch:
print 'Skip shutting down chart screen'
else:
print 'Shutting down chart screen: ', chart_host_id
screen_id_arg = ('screen=%s' % chart_host_id)
cmd = ['python', os.path.join(os.environ['CAMERA_ITS_TOP'], 'tools',
'turn_off_screen.py'), screen_id_arg]
screen_off_code = subprocess.call(cmd)
assert screen_off_code == 0
print 'Shutting down DUT screen: ', device_id
screen_id_arg = ('screen=%s' % device_id)
cmd = ['python', os.path.join(os.environ['CAMERA_ITS_TOP'], 'tools',
'turn_off_screen.py'), screen_id_arg]
screen_off_code = subprocess.call(cmd)
assert screen_off_code == 0
print "ITS tests finished. Please go back to CtsVerifier and proceed"
if __name__ == '__main__':
main()