1# Copyright 2014 The Android Open Source Project 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14 15import copy 16import math 17import os 18import os.path 19import re 20import subprocess 21import sys 22import tempfile 23import threading 24import time 25 26import its.caps 27import its.cv2image 28import its.device 29from its.device import ItsSession 30import its.image 31 32import numpy as np 33 34# For sanity checking the installed APK's target SDK version 35MIN_SUPPORTED_SDK_VERSION = 28 # P 36 37CHART_DELAY = 1 # seconds 38CHART_DISTANCE = 30.0 # cm 39CHART_HEIGHT = 13.5 # cm 40CHART_LEVEL = 96 41CHART_SCALE_START = 0.65 42CHART_SCALE_STOP = 1.35 43CHART_SCALE_STEP = 0.025 44FACING_EXTERNAL = 2 45NUM_TRYS = 2 46PROC_TIMEOUT_CODE = -101 # terminated process return -process_id 47PROC_TIMEOUT_TIME = 900 # timeout in seconds for a process (15 minutes) 48SCENE3_FILE = os.path.join(os.environ['CAMERA_ITS_TOP'], 'pymodules', 'its', 49 'test_images', 'ISO12233.png') 50SKIP_RET_CODE = 101 # note this must be same as tests/scene*/test_* 51VGA_HEIGHT = 480 52VGA_WIDTH = 640 53 54# Not yet mandated tests 55NOT_YET_MANDATED = { 56 'scene0': [ 57 'test_test_patterns', 58 'test_tonemap_curve' 59 ], 60 'scene1': [ 61 'test_ae_precapture_trigger', 62 'test_channel_saturation' 63 ], 64 'scene2': [ 65 'test_auto_per_frame_control' 66 ], 67 'scene2b': [], 68 'scene2c': [], 69 'scene3': [], 70 'scene4': [], 71 'scene5': [], 72 'sensor_fusion': [] 73} 74 75# Must match mHiddenPhysicalCameraSceneIds in ItsTestActivity.java 76HIDDEN_PHYSICAL_CAMERA_TESTS = { 77 'scene0': [ 78 'test_burst_capture', 79 'test_metadata', 80 'test_read_write', 81 'test_sensor_events' 82 ], 83 'scene1': [ 84 'test_exposure', 85 'test_dng_noise_model', 86 'test_linearity', 87 'test_raw_exposure', 88 'test_raw_sensitivity' 89 ], 90 'scene2': [ 91 'test_faces', 92 'test_num_faces' 93 ], 94 'scene2b': [], 95 'scene2c': [], 96 'scene3': [], 97 'scene4': [ 98 'test_aspect_ratio_and_crop' 99 ], 100 'scene5': [], 101 'sensor_fusion': [ 102 'test_sensor_fusion' 103 ] 104} 105 106def run_subprocess_with_timeout(cmd, fout, ferr, outdir): 107 """Run subprocess with a timeout. 108 109 Args: 110 cmd: list containing python command 111 fout: stdout file for the test 112 ferr: stderr file for the test 113 outdir: dir location for fout/ferr 114 115 Returns: 116 process status or PROC_TIMEOUT_CODE if timer maxes 117 """ 118 119 proc = subprocess.Popen( 120 cmd, stdout=fout, stderr=ferr, cwd=outdir) 121 timer = threading.Timer(PROC_TIMEOUT_TIME, proc.kill) 122 123 try: 124 timer.start() 125 proc.communicate() 126 test_code = proc.returncode 127 finally: 128 timer.cancel() 129 130 if test_code < 0: 131 return PROC_TIMEOUT_CODE 132 else: 133 return test_code 134 135 136def calc_camera_fov(camera_id, hidden_physical_id): 137 """Determine the camera field of view from internal params.""" 138 with ItsSession(camera_id, hidden_physical_id) as cam: 139 props = cam.get_camera_properties() 140 props = cam.override_with_hidden_physical_camera_props(props) 141 focal_ls = props['android.lens.info.availableFocalLengths'] 142 if len(focal_ls) > 1: 143 print 'Doing capture to determine logical camera focal length' 144 cap = cam.do_capture(its.objects.auto_capture_request()) 145 focal_l = cap['metadata']['android.lens.focalLength'] 146 else: 147 focal_l = focal_ls[0] 148 sensor_size = props['android.sensor.info.physicalSize'] 149 diag = math.sqrt(sensor_size['height'] ** 2 + 150 sensor_size['width'] ** 2) 151 try: 152 fov = str(round(2 * math.degrees(math.atan(diag / (2 * focal_l))), 2)) 153 except ValueError: 154 fov = str(0) 155 print 'Calculated FoV: %s' % fov 156 return fov 157 158 159def evaluate_socket_failure(err_file_path): 160 """Determine if test fails due to socket FAIL.""" 161 socket_fail = False 162 with open(err_file_path, 'r') as ferr: 163 for line in ferr: 164 if (line.find('socket.error') != -1 or 165 line.find('socket.timeout') != -1 or 166 line.find('Problem with socket') != -1): 167 socket_fail = True 168 return socket_fail 169 170 171def skip_sensor_fusion(camera_id): 172 """Determine if sensor fusion test is skipped for this camera.""" 173 174 skip_code = SKIP_RET_CODE 175 with ItsSession(camera_id) as cam: 176 props = cam.get_camera_properties() 177 if (its.caps.sensor_fusion(props) and its.caps.manual_sensor(props) and 178 props['android.lens.facing'] is not FACING_EXTERNAL): 179 skip_code = None 180 return skip_code 181 182 183def main(): 184 """Run all the automated tests, saving intermediate files, and producing 185 a summary/report of the results. 186 187 Script should be run from the top-level CameraITS directory. 188 189 Command line arguments: 190 camera: the camera(s) to be tested. Use comma to separate multiple 191 camera Ids. Ex: "camera=0,1" or "camera=1" 192 device: device id for adb 193 scenes: the test scene(s) to be executed. Use comma to separate 194 multiple scenes. Ex: "scenes=scene0,scene1" or 195 "scenes=0,1,sensor_fusion" (sceneX can be abbreviated by X 196 where X is a integer) 197 chart: [Experimental] another android device served as test chart 198 display. When this argument presents, change of test scene 199 will be handled automatically. Note that this argument 200 requires special physical/hardware setup to work and may not 201 work on all android devices. 202 result: Device ID to forward results to (in addition to the device 203 that the tests are running on). 204 rot_rig: [Experimental] ID of the rotation rig being used (formatted as 205 "<vendor ID>:<product ID>:<channel #>" or "default") 206 tmp_dir: location of temp directory for output files 207 skip_scene_validation: force skip scene validation. Used when test scene 208 is setup up front and don't require tester validation. 209 dist: [Experimental] chart distance in cm. 210 """ 211 212 all_scenes = ["scene0", "scene1", "scene2", "scene2b", "scene2c", "scene3", "scene4", "scene5", 213 "sensor_fusion"] 214 215 auto_scenes = ["scene0", "scene1", "scene2", "scene2b", "scene2c", "scene3", "scene4"] 216 217 scene_req = { 218 "scene0": None, 219 "scene1": "A grey card covering at least the middle 30% of the scene", 220 "scene2": "A picture containing human faces", 221 "scene2b": "A picture containing human faces", 222 "scene2c": "A picture containing human faces", 223 "scene3": "The ISO 12233 chart", 224 "scene4": "A specific test page of a circle covering at least the " 225 "middle 50% of the scene. See CameraITS.pdf section 2.3.4 " 226 "for more details", 227 "scene5": "Capture images with a diffuser attached to the camera. See " 228 "CameraITS.pdf section 2.3.4 for more details", 229 "sensor_fusion": "Rotating checkboard pattern. See " 230 "sensor_fusion/SensorFusion.pdf for detailed " 231 "instructions.\nNote that this test will be skipped " 232 "on devices not supporting REALTIME camera timestamp." 233 } 234 scene_extra_args = { 235 "scene5": ["doAF=False"] 236 } 237 238 camera_id_combos = [] 239 scenes = [] 240 chart_host_id = None 241 result_device_id = None 242 rot_rig_id = None 243 tmp_dir = None 244 skip_scene_validation = False 245 chart_distance = CHART_DISTANCE 246 chart_level = CHART_LEVEL 247 one_camera_argv = sys.argv[1:] 248 249 for s in list(sys.argv[1:]): 250 if s[:7] == "camera=" and len(s) > 7: 251 camera_ids = s[7:].split(',') 252 camera_id_combos = its.device.parse_camera_ids(camera_ids) 253 one_camera_argv.remove(s) 254 elif s[:7] == "scenes=" and len(s) > 7: 255 scenes = s[7:].split(',') 256 elif s[:6] == 'chart=' and len(s) > 6: 257 chart_host_id = s[6:] 258 elif s[:7] == 'result=' and len(s) > 7: 259 result_device_id = s[7:] 260 elif s[:8] == 'rot_rig=' and len(s) > 8: 261 rot_rig_id = s[8:] # valid values: 'default' or '$VID:$PID:$CH' 262 # The default '$VID:$PID:$CH' is '04d8:fc73:1' 263 elif s[:8] == 'tmp_dir=' and len(s) > 8: 264 tmp_dir = s[8:] 265 elif s == 'skip_scene_validation': 266 skip_scene_validation = True 267 elif s[:5] == 'dist=' and len(s) > 5: 268 chart_distance = float(re.sub('cm', '', s[5:])) 269 elif s[:11] == 'brightness=' and len(s) > 11: 270 chart_level = s[11:] 271 272 chart_dist_arg = 'dist= ' + str(chart_distance) 273 chart_level_arg = 'brightness=' + str(chart_level) 274 auto_scene_switch = chart_host_id is not None 275 merge_result_switch = result_device_id is not None 276 277 # Run through all scenes if user does not supply one 278 possible_scenes = auto_scenes if auto_scene_switch else all_scenes 279 if not scenes: 280 scenes = possible_scenes 281 else: 282 # Validate user input scene names 283 valid_scenes = True 284 temp_scenes = [] 285 for s in scenes: 286 if s in possible_scenes: 287 temp_scenes.append(s) 288 else: 289 try: 290 # Try replace "X" to "sceneX" 291 scene_str = "scene" + s 292 if scene_str not in possible_scenes: 293 valid_scenes = False 294 break 295 temp_scenes.append(scene_str) 296 except ValueError: 297 valid_scenes = False 298 break 299 300 if not valid_scenes: 301 print 'Unknown scene specified:', s 302 assert False 303 scenes = temp_scenes 304 305 # Initialize test results 306 results = {} 307 result_key = ItsSession.RESULT_KEY 308 for s in all_scenes: 309 results[s] = {result_key: ItsSession.RESULT_NOT_EXECUTED} 310 311 # Make output directories to hold the generated files. 312 topdir = tempfile.mkdtemp(dir=tmp_dir) 313 subprocess.call(['chmod', 'g+rx', topdir]) 314 print "Saving output files to:", topdir, "\n" 315 316 device_id = its.device.get_device_id() 317 device_id_arg = "device=" + device_id 318 print "Testing device " + device_id 319 320 # Sanity check CtsVerifier SDK level 321 # Here we only do warning as there is no guarantee on pm dump output formt not changed 322 # Also sometimes it's intentional to run mismatched versions 323 cmd = "adb -s %s shell pm dump com.android.cts.verifier" % (device_id) 324 dump_path = os.path.join(topdir, 'CtsVerifier.txt') 325 with open(dump_path, 'w') as fout: 326 fout.write('ITS minimum supported SDK version is %d\n--\n' % (MIN_SUPPORTED_SDK_VERSION)) 327 fout.flush() 328 ret_code = subprocess.call(cmd.split(), stdout=fout) 329 330 if ret_code != 0: 331 print "Warning: cannot get CtsVerifier SDK version. Is CtsVerifier installed?" 332 333 ctsv_version = None 334 ctsv_version_name = None 335 with open(dump_path, 'r') as f: 336 target_sdk_found = False 337 version_name_found = False 338 for line in f: 339 match = re.search('targetSdk=([0-9]+)', line) 340 if match: 341 ctsv_version = int(match.group(1)) 342 target_sdk_found = True 343 match = re.search('versionName=([\S]+)$', line) 344 if match: 345 ctsv_version_name = match.group(1) 346 version_name_found = True 347 if target_sdk_found and version_name_found: 348 break 349 350 if ctsv_version is None: 351 print "Warning: cannot get CtsVerifier SDK version. Is CtsVerifier installed?" 352 elif ctsv_version < MIN_SUPPORTED_SDK_VERSION: 353 print "Warning: CtsVerifier version (%d) < ITS version (%d), is this intentional?" % ( 354 ctsv_version, MIN_SUPPORTED_SDK_VERSION) 355 else: 356 print "CtsVerifier targetSdk is", ctsv_version 357 if ctsv_version_name: 358 print "CtsVerifier version name is", ctsv_version_name 359 360 # Hard check on ItsService/host script version that should catch incompatible APK/script 361 with ItsSession() as cam: 362 cam.check_its_version_compatible() 363 364 # Sanity Check for devices 365 device_bfp = its.device.get_device_fingerprint(device_id) 366 assert device_bfp is not None 367 368 if auto_scene_switch: 369 chart_host_bfp = its.device.get_device_fingerprint(chart_host_id) 370 assert chart_host_bfp is not None 371 372 if merge_result_switch: 373 result_device_bfp = its.device.get_device_fingerprint(result_device_id) 374 assert_err_msg = ('Cannot merge result to a different build, from ' 375 '%s to %s' % (device_bfp, result_device_bfp)) 376 assert device_bfp == result_device_bfp, assert_err_msg 377 378 # user doesn't specify camera id, run through all cameras 379 if not camera_id_combos: 380 with its.device.ItsSession() as cam: 381 camera_ids = cam.get_camera_ids() 382 camera_id_combos = its.device.parse_camera_ids(camera_ids); 383 384 print "Running ITS on camera: %s, scene %s" % (camera_id_combos, scenes) 385 386 if auto_scene_switch: 387 # merge_result only supports run_parallel_tests 388 if merge_result_switch and camera_ids[0] == "1": 389 print "Skip chart screen" 390 time.sleep(1) 391 else: 392 print "Waking up chart screen: ", chart_host_id 393 screen_id_arg = ("screen=%s" % chart_host_id) 394 cmd = ["python", os.path.join(os.environ["CAMERA_ITS_TOP"], "tools", 395 "wake_up_screen.py"), screen_id_arg, 396 chart_level_arg] 397 wake_code = subprocess.call(cmd) 398 assert wake_code == 0 399 400 for id_combo in camera_id_combos: 401 camera_fov = calc_camera_fov(id_combo.id, id_combo.sub_id) 402 id_combo_string = id_combo.id; 403 has_hidden_sub_camera = id_combo.sub_id is not None 404 if has_hidden_sub_camera: 405 id_combo_string += ":" + id_combo.sub_id 406 scenes = [scene for scene in scenes if HIDDEN_PHYSICAL_CAMERA_TESTS[scene]] 407 # Loop capturing images until user confirm test scene is correct 408 camera_id_arg = "camera=" + id_combo.id 409 print "Preparing to run ITS on camera", id_combo_string, "for scenes ", scenes 410 411 os.mkdir(os.path.join(topdir, id_combo_string)) 412 for d in scenes: 413 os.mkdir(os.path.join(topdir, id_combo_string, d)) 414 415 tot_tests = [] 416 tot_pass = 0 417 for scene in scenes: 418 skip_code = None 419 tests = [(s[:-3], os.path.join("tests", scene, s)) 420 for s in os.listdir(os.path.join("tests", scene)) 421 if s[-3:] == ".py" and s[:4] == "test"] 422 tests.sort() 423 tot_tests.extend(tests) 424 425 summary = "Cam" + id_combo_string + " " + scene + "\n" 426 numpass = 0 427 numskip = 0 428 num_not_mandated_fail = 0 429 numfail = 0 430 validate_switch = True 431 if scene_req[scene] is not None: 432 out_path = os.path.join(topdir, id_combo_string, scene+".jpg") 433 out_arg = "out=" + out_path 434 if scene == 'sensor_fusion': 435 skip_code = skip_sensor_fusion(id_combo.id) 436 if rot_rig_id or skip_code == SKIP_RET_CODE: 437 validate_switch = False 438 if skip_scene_validation: 439 validate_switch = False 440 cmd = None 441 if auto_scene_switch: 442 if (not merge_result_switch or 443 (merge_result_switch and id_combo_string == '0')): 444 scene_arg = 'scene=' + scene 445 fov_arg = 'fov=' + camera_fov 446 cmd = ['python', 447 os.path.join(os.getcwd(), 'tools/load_scene.py'), 448 scene_arg, chart_dist_arg, fov_arg, screen_id_arg] 449 else: 450 time.sleep(CHART_DELAY) 451 else: 452 # Skip scene validation under certain conditions 453 if validate_switch and not merge_result_switch: 454 scene_arg = 'scene=' + scene_req[scene] 455 extra_args = scene_extra_args.get(scene, []) 456 cmd = ['python', 457 os.path.join(os.getcwd(), 458 'tools/validate_scene.py'), 459 camera_id_arg, out_arg, 460 scene_arg, device_id_arg] + extra_args 461 if cmd is not None: 462 valid_scene_code = subprocess.call(cmd, cwd=topdir) 463 assert valid_scene_code == 0 464 print 'Start running ITS on camera %s, %s' % (id_combo_string, scene) 465 # Extract chart from scene for scene3 once up front 466 chart_loc_arg = '' 467 chart_height = CHART_HEIGHT 468 if scene == 'scene3': 469 chart_height *= its.cv2image.calc_chart_scaling( 470 chart_distance, camera_fov) 471 chart = its.cv2image.Chart(SCENE3_FILE, chart_height, 472 chart_distance, CHART_SCALE_START, 473 CHART_SCALE_STOP, CHART_SCALE_STEP, 474 id_combo.id) 475 chart_loc_arg = 'chart_loc=%.2f,%.2f,%.2f,%.2f,%.3f' % ( 476 chart.xnorm, chart.ynorm, chart.wnorm, chart.hnorm, 477 chart.scale) 478 # Run each test, capturing stdout and stderr. 479 for (testname, testpath) in tests: 480 # Only pick predefined tests for hidden physical camera 481 if has_hidden_sub_camera and \ 482 testname not in HIDDEN_PHYSICAL_CAMERA_TESTS[scene]: 483 numskip += 1 484 continue 485 if auto_scene_switch: 486 if merge_result_switch and id_combo_string == '0': 487 # Send an input event to keep the screen not dimmed. 488 # Since we are not using camera of chart screen, FOCUS event 489 # should do nothing but keep the screen from dimming. 490 # The "sleep after x minutes of inactivity" display setting 491 # determines how long this command can keep screen bright. 492 # Setting it to something like 30 minutes should be enough. 493 cmd = ('adb -s %s shell input keyevent FOCUS' 494 % chart_host_id) 495 subprocess.call(cmd.split()) 496 t0 = time.time() 497 for num_try in range(NUM_TRYS): 498 outdir = os.path.join(topdir, id_combo_string, scene) 499 outpath = os.path.join(outdir, testname+'_stdout.txt') 500 errpath = os.path.join(outdir, testname+'_stderr.txt') 501 if scene == 'sensor_fusion': 502 if skip_code is not SKIP_RET_CODE: 503 if rot_rig_id: 504 print 'Rotating phone w/ rig %s' % rot_rig_id 505 rig = ('python tools/rotation_rig.py rotator=%s' % 506 rot_rig_id) 507 subprocess.Popen(rig.split()) 508 else: 509 print 'Rotate phone 15s as shown in SensorFusion.pdf' 510 else: 511 test_code = skip_code 512 if skip_code is not SKIP_RET_CODE: 513 cmd = ['python', os.path.join(os.getcwd(), testpath)] 514 cmd += one_camera_argv + ["camera="+id_combo_string] + [chart_loc_arg] 515 cmd += [chart_dist_arg] 516 with open(outpath, 'w') as fout, open(errpath, 'w') as ferr: 517 test_code = run_subprocess_with_timeout( 518 cmd, fout, ferr, outdir) 519 if test_code == 0 or test_code == SKIP_RET_CODE: 520 break 521 else: 522 socket_fail = evaluate_socket_failure(errpath) 523 if socket_fail or test_code == PROC_TIMEOUT_CODE: 524 if num_try != NUM_TRYS-1: 525 print ' Retry %s/%s' % (scene, testname) 526 else: 527 break 528 else: 529 break 530 t1 = time.time() 531 532 test_failed = False 533 if test_code == 0: 534 retstr = "PASS " 535 numpass += 1 536 elif test_code == SKIP_RET_CODE: 537 retstr = "SKIP " 538 numskip += 1 539 elif test_code != 0 and testname in NOT_YET_MANDATED[scene]: 540 retstr = "FAIL*" 541 num_not_mandated_fail += 1 542 else: 543 retstr = "FAIL " 544 numfail += 1 545 test_failed = True 546 547 msg = "%s %s/%s [%.1fs]" % (retstr, scene, testname, t1-t0) 548 print msg 549 its.device.adb_log(device_id, msg) 550 msg_short = "%s %s [%.1fs]" % (retstr, testname, t1-t0) 551 if test_failed: 552 summary += msg_short + "\n" 553 554 if numskip > 0: 555 skipstr = ", %d test%s skipped" % ( 556 numskip, "s" if numskip > 1 else "") 557 else: 558 skipstr = "" 559 560 test_result = "\n%d / %d tests passed (%.1f%%)%s" % ( 561 numpass + num_not_mandated_fail, len(tests) - numskip, 562 100.0 * float(numpass + num_not_mandated_fail) / 563 (len(tests) - numskip) 564 if len(tests) != numskip else 100.0, skipstr) 565 print test_result 566 567 if num_not_mandated_fail > 0: 568 msg = "(*) tests are not yet mandated" 569 print msg 570 571 tot_pass += numpass 572 print "%s compatibility score: %.f/100\n" % ( 573 scene, 100.0 * numpass / len(tests)) 574 575 summary_path = os.path.join(topdir, id_combo_string, scene, "summary.txt") 576 with open(summary_path, "w") as f: 577 f.write(summary) 578 579 passed = numfail == 0 580 results[scene][result_key] = (ItsSession.RESULT_PASS if passed 581 else ItsSession.RESULT_FAIL) 582 results[scene][ItsSession.SUMMARY_KEY] = summary_path 583 584 if tot_tests: 585 print "Compatibility Score: %.f/100" % (100.0 * tot_pass / len(tot_tests)) 586 else: 587 print "Compatibility Score: 0/100" 588 589 msg = "Reporting ITS result to CtsVerifier" 590 print msg 591 its.device.adb_log(device_id, msg) 592 if merge_result_switch: 593 # results are modified by report_result 594 results_backup = copy.deepcopy(results) 595 its.device.report_result(result_device_id, id_combo_string, results_backup) 596 597 # Report hidden_physical_id results as well. 598 its.device.report_result(device_id, id_combo_string, results) 599 600 if auto_scene_switch: 601 if merge_result_switch: 602 print 'Skip shutting down chart screen' 603 else: 604 print 'Shutting down chart screen: ', chart_host_id 605 screen_id_arg = ('screen=%s' % chart_host_id) 606 cmd = ['python', os.path.join(os.environ['CAMERA_ITS_TOP'], 'tools', 607 'turn_off_screen.py'), screen_id_arg] 608 screen_off_code = subprocess.call(cmd) 609 assert screen_off_code == 0 610 611 print 'Shutting down DUT screen: ', device_id 612 screen_id_arg = ('screen=%s' % device_id) 613 cmd = ['python', os.path.join(os.environ['CAMERA_ITS_TOP'], 'tools', 614 'turn_off_screen.py'), screen_id_arg] 615 screen_off_code = subprocess.call(cmd) 616 assert screen_off_code == 0 617 618 print "ITS tests finished. Please go back to CtsVerifier and proceed" 619 620if __name__ == '__main__': 621 main() 622