| # 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 json |
| import logging |
| import os |
| import os.path |
| import subprocess |
| import sys |
| import tempfile |
| import time |
| import yaml |
| |
| import capture_request_utils |
| import image_processing_utils |
| import its_session_utils |
| |
| YAML_FILE_DIR = os.environ['CAMERA_ITS_TOP'] |
| CONFIG_FILE = os.path.join(YAML_FILE_DIR, 'config.yml') |
| TEST_KEY_TABLET = 'tablet' |
| TEST_KEY_SENSOR_FUSION = 'sensor_fusion' |
| LOAD_SCENE_DELAY = 1 # seconds |
| ACTIVITY_START_WAIT = 1.5 # seconds |
| |
| RESULT_PASS = 'PASS' |
| RESULT_FAIL = 'FAIL' |
| RESULT_NOT_EXECUTED = 'NOT_EXECUTED' |
| RESULT_KEY = 'result' |
| SUMMARY_KEY = 'summary' |
| RESULT_VALUES = {RESULT_PASS, RESULT_FAIL, RESULT_NOT_EXECUTED} |
| ITS_TEST_ACTIVITY = 'com.android.cts.verifier/.camera.its.ItsTestActivity' |
| ACTION_ITS_RESULT = 'com.android.cts.verifier.camera.its.ACTION_ITS_RESULT' |
| EXTRA_VERSION = 'camera.its.extra.VERSION' |
| CURRENT_ITS_VERSION = '1.0' # version number to sync with CtsVerifier |
| EXTRA_CAMERA_ID = 'camera.its.extra.CAMERA_ID' |
| EXTRA_RESULTS = 'camera.its.extra.RESULTS' |
| TIME_KEY_START = 'start' |
| TIME_KEY_END = 'end' |
| |
| # All possible scenes |
| # Notes on scene names: |
| # scene*_1/2/... are same scene split to load balance run times for scenes |
| # scene*_a/b/... are similar scenes that share one or more tests |
| _ALL_SCENES = [ |
| 'scene0', 'scene1_1', 'scene1_2', 'scene2_a', 'scene2_b', 'scene2_c', |
| 'scene2_d', 'scene2_e', 'scene3', 'scene4', 'scene5', 'scene6', |
| 'sensor_fusion', 'scene_change' |
| ] |
| |
| # Scenes that can be automated through tablet display |
| _AUTO_SCENES = [ |
| 'scene0', 'scene1_1', 'scene1_2', 'scene2_a', 'scene2_b', 'scene2_c', |
| 'scene2_d', 'scene2_e', 'scene3', 'scene4', 'scene6', 'scene_change' |
| ] |
| |
| # Scenes that are logically grouped and can be called as group |
| _GROUPED_SCENES = { |
| 'scene1': ['scene1_1', 'scene1_2'], |
| 'scene2': ['scene2_a', 'scene2_b', 'scene2_c', 'scene2_d', 'scene2_e'] |
| } |
| |
| # Tests run in more than 1 scene. |
| # List is created of type ['scene_source', 'test_to_be_repeated'] |
| # for the test run in current scene. |
| _REPEATED_TESTS = { |
| 'scene0': [], |
| 'scene1_1': [], |
| 'scene1_2': [], |
| 'scene2_a': [], |
| 'scene2_b': [['scene2_a', 'test_num_faces']], |
| 'scene2_c': [['scene2_a', 'test_num_faces']], |
| 'scene2_d': [['scene2_a', 'test_num_faces']], |
| 'scene2_e': [['scene2_a', 'test_num_faces']], |
| 'scene3': [], |
| 'scene4': [], |
| 'scene5': [], |
| 'scene6': [], |
| 'sensor_fusion': [], |
| 'scene_change': [] |
| } |
| |
| # Scene requirements for manual testing. |
| _SCENE_REQ = { |
| 'scene0': None, |
| 'scene1_1': 'A grey card covering at least the middle 30% of the scene', |
| 'scene1_2': 'A grey card covering at least the middle 30% of the scene', |
| 'scene2_a': 'The picture with 3 faces in tests/scene2_a/scene2_a.pdf', |
| 'scene2_b': 'The picture with 3 faces in tests/scene2_b/scene2_b.pdf', |
| 'scene2_c': 'The picture with 3 faces in tests/scene2_c/scene2_c.pdf', |
| 'scene2_d': 'The picture with 3 faces in tests/scene2_d/scene2_d.pdf', |
| 'scene2_e': 'The picture with 3 faces in tests/scene2_e/scene2_e.pdf', |
| 'scene3': 'The ISO12233 chart', |
| 'scene4': 'A test chart of a circle covering at least the middle 50% of ' |
| 'the scene. See tests/scene4/scene4.pdf', |
| 'scene5': 'Capture images with a diffuser attached to the camera. ' |
| 'See CameraITS.pdf section 2.3.4 for more details', |
| 'scene6': 'A grid of black circles on a white background. ' |
| 'See tests/scene6/scene6.pdf', |
| 'sensor_fusion': 'A checkerboard pattern for phone to rotate in front of ' |
| 'in tests/sensor_fusion/checkerboard.pdf\n' |
| 'See tests/sensor_fusion/SensorFusion.pdf for detailed ' |
| 'instructions.\nNote that this test will be skipped ' |
| 'on devices not supporting REALTIME camera timestamp.', |
| 'scene_change': 'The picture with 3 faces in tests/scene2_e/scene2_e.pdf', |
| } |
| |
| |
| SUB_CAMERA_TESTS = { |
| 'scene0': [ |
| 'test_burst_capture', |
| 'test_metadata', |
| 'test_read_write', |
| 'test_sensor_events', |
| 'test_solid_color_test_pattern', |
| 'test_unified_timestamps', |
| ], |
| 'scene1_1': [ |
| 'test_exposure', |
| 'test_dng_noise_model', |
| 'test_linearity', |
| ], |
| 'scene1_2': [ |
| 'test_raw_exposure', |
| 'test_raw_sensitivity', |
| ], |
| 'scene2_a': [ |
| 'test_faces', |
| 'test_num_faces', |
| ], |
| 'scene4': [ |
| 'test_aspect_ratio_and_crop', |
| ], |
| 'sensor_fusion': [ |
| 'test_sensor_fusion', |
| ], |
| } |
| |
| _DST_SCENE_DIR = '/mnt/sdcard/Download/' |
| MOBLY_TEST_SUMMARY_TXT_FILE = 'test_mobly_summary.txt' |
| |
| |
| def run(cmd): |
| """Replaces os.system call, while hiding stdout+stderr messages.""" |
| with open(os.devnull, 'wb') as devnull: |
| subprocess.check_call(cmd.split(), stdout=devnull, stderr=subprocess.STDOUT) |
| |
| |
| def report_result(device_id, camera_id, results): |
| """Sends a pass/fail result to the device, via an intent. |
| |
| Args: |
| device_id: The ID string of the device to report the results to. |
| camera_id: The ID string of the camera for which to report pass/fail. |
| results: a dictionary contains all ITS scenes as key and result/summary of |
| current ITS run. See test_report_result unit test for an example. |
| """ |
| adb = f'adb -s {device_id}' |
| |
| # Start ItsTestActivity to receive test results |
| cmd = f'{adb} shell am start {ITS_TEST_ACTIVITY} --activity-brought-to-front' |
| run(cmd) |
| time.sleep(ACTIVITY_START_WAIT) |
| |
| # Validate/process results argument |
| for scene in results: |
| if RESULT_KEY not in results[scene]: |
| raise ValueError(f'ITS result not found for {scene}') |
| if results[scene][RESULT_KEY] not in RESULT_VALUES: |
| raise ValueError(f'Unknown ITS result for {scene}: {results[RESULT_KEY]}') |
| if SUMMARY_KEY in results[scene]: |
| device_summary_path = f'/sdcard/its_camera{camera_id}_{scene}.txt' |
| run('%s push %s %s' % |
| (adb, results[scene][SUMMARY_KEY], device_summary_path)) |
| results[scene][SUMMARY_KEY] = device_summary_path |
| |
| json_results = json.dumps(results) |
| cmd = (f"{adb} shell am broadcast -a {ACTION_ITS_RESULT} --es {EXTRA_VERSION}" |
| f" {CURRENT_ITS_VERSION} --es {EXTRA_CAMERA_ID} {camera_id} --es " |
| f"{EXTRA_RESULTS} \'{json_results}\'") |
| if len(cmd) > 4095: |
| logging.info('ITS command string might be too long! len:%s', len(cmd)) |
| run(cmd) |
| |
| |
| def load_scenes_on_tablet(scene, tablet_id): |
| """Copies scenes onto the tablet before running the tests. |
| |
| Args: |
| scene: Name of the scene to copy image files. |
| tablet_id: adb id of tablet |
| """ |
| logging.info('Copying files to tablet: %s', tablet_id) |
| scene_dir = os.listdir( |
| os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests', scene)) |
| for file_name in scene_dir: |
| if file_name.endswith('.pdf'): |
| src_scene_file = os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests', |
| scene, file_name) |
| cmd = f'adb -s {tablet_id} push {src_scene_file} {_DST_SCENE_DIR}' |
| subprocess.Popen(cmd.split()) |
| time.sleep(LOAD_SCENE_DELAY) |
| logging.info('Finished copying files to tablet.') |
| |
| |
| def check_manual_scenes(device_id, camera_id, scene, out_path): |
| """Halt run to change scenes. |
| |
| Args: |
| device_id: id of device |
| camera_id: id of camera |
| scene: Name of the scene to copy image files. |
| out_path: output file location |
| """ |
| logging.info('No chart tablet defined. Manual testing.') |
| with its_session_utils.ItsSession( |
| device_id=device_id, |
| camera_id=camera_id) as cam: |
| props = cam.get_camera_properties() |
| |
| while True: |
| input(f'\n Press <ENTER> after positioning camera {camera_id} with ' |
| f'{scene}.\n The scene setup should be: \n {_SCENE_REQ[scene]}\n') |
| # Converge 3A prior to capture. |
| cam.do_3a() |
| req, fmt = capture_request_utils.get_fastest_auto_capture_settings(props) |
| logging.info('Capturing an image to check the test scene') |
| cap = cam.do_capture(req, fmt) |
| img = image_processing_utils.convert_capture_to_rgb_image(cap) |
| img_name = os.path.join(out_path, f'test_{scene}.jpg') |
| logging.info('Please check scene setup in %s', img_name) |
| image_processing_utils.write_image(img, img_name) |
| choice = input('Is the image okay for ITS {scene}? (Y/N)').lower() |
| if choice == 'y': |
| break |
| |
| |
| def get_config_file_contents(): |
| """Read the config file contents from a YML file. |
| |
| Args: |
| None |
| |
| Returns: |
| config_file_contents: a dict read from config.yml |
| """ |
| with open(CONFIG_FILE) as file: |
| config_file_contents = yaml.load(file, yaml.FullLoader) |
| return config_file_contents |
| |
| |
| def get_test_params(config_file_contents): |
| """Reads the config file parameters. |
| |
| Args: |
| config_file_contents: dict read from config.yml file |
| |
| Returns: |
| dict of test parameters |
| """ |
| test_params = None |
| for _, j in config_file_contents.items(): |
| for datadict in j: |
| test_params = datadict.get('TestParams') |
| return test_params |
| |
| |
| def get_device_serial_number(device, config_file_contents): |
| """Returns the serial number of the device with label from the config file. |
| |
| The config file contains TestBeds dictionary which contains Controllers and |
| Android Device dicts.The two devices used by the test per box are listed |
| here labels dut and tablet. Parse through the nested TestBeds dict to get |
| the Android device details. |
| |
| Args: |
| device: String device label as specified in config file.dut/tablet |
| config_file_contents: dict read from config.yml file |
| """ |
| |
| for _, j in config_file_contents.items(): |
| for datadict in j: |
| android_device_contents = datadict.get('Controllers') |
| for device_dict in android_device_contents.get('AndroidDevice'): |
| for _, label in device_dict.items(): |
| if label == 'tablet': |
| tablet_device_id = device_dict.get('serial') |
| if label == 'dut': |
| dut_device_id = device_dict.get('serial') |
| if device == 'tablet': |
| return tablet_device_id |
| else: |
| return dut_device_id |
| |
| |
| def expand_scene(scene, scenes): |
| """Expand a grouped scene and append its sub_scenes to scenes. |
| |
| Args: |
| scene: scene in GROUPED_SCENES dict |
| scenes: list of scenes to append to |
| |
| Returns: |
| updated scenes |
| """ |
| logging.info('Expanding %s to %s.', scene, str(_GROUPED_SCENES[scene])) |
| for sub_scene in _GROUPED_SCENES[scene]: |
| scenes.append(sub_scene) |
| |
| |
| def get_updated_yml_file(yml_file_contents): |
| """Create a new yml file and write the testbed contents in it. |
| |
| This testbed file is per box and contains all the parameters and |
| device id used by the mobly tests. |
| |
| Args: |
| yml_file_contents: Data to write in yml file. |
| |
| Returns: |
| Updated yml file contents. |
| """ |
| os.chmod(YAML_FILE_DIR, 0o755) |
| _, new_yaml_file = tempfile.mkstemp( |
| suffix='.yml', prefix='config_', dir=YAML_FILE_DIR) |
| with open(new_yaml_file, 'w') as f: |
| yaml.dump(yml_file_contents, stream=f, default_flow_style=False) |
| new_yaml_file_name = os.path.basename(new_yaml_file) |
| return new_yaml_file_name |
| |
| |
| def main(): |
| """Run all the Camera ITS automated tests. |
| |
| 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" |
| scenes: the test scene(s) to be executed. Use comma to separate |
| multiple scenes. Ex: "scenes=scene0,scene1_1" or |
| "scenes=0,1_1,sensor_fusion" (sceneX can be abbreviated by X |
| where X is scene name minus 'scene') |
| """ |
| logging.basicConfig(level=logging.INFO) |
| # Make output directories to hold the generated files. |
| topdir = tempfile.mkdtemp(prefix='CameraITS_') |
| subprocess.call(['chmod', 'g+rx', topdir]) |
| logging.info('Saving output files to: %s', topdir) |
| |
| scenes = [] |
| camera_id_combos = [] |
| # Override camera & scenes with cmd line values if available |
| for s in list(sys.argv[1:]): |
| if 'scenes=' in s: |
| scenes = s.split('=')[1].split(',') |
| elif 'camera=' in s: |
| camera_id_combos = s.split('=')[1].split(',') |
| |
| # Read config file and extract relevant TestBed |
| config_file_contents = get_config_file_contents() |
| for i in config_file_contents['TestBeds']: |
| if scenes == ['sensor_fusion']: |
| if TEST_KEY_SENSOR_FUSION not in i['Name'].lower(): |
| config_file_contents['TestBeds'].remove(i) |
| else: |
| if TEST_KEY_SENSOR_FUSION in i['Name'].lower(): |
| config_file_contents['TestBeds'].remove(i) |
| |
| # Get test parameters from config file |
| test_params_content = get_test_params(config_file_contents) |
| if not camera_id_combos: |
| camera_id_combos = str(test_params_content['camera']).split(',') |
| if not scenes: |
| scenes = test_params_content['scene'].split(',') |
| |
| device_id = get_device_serial_number('dut', config_file_contents) |
| |
| if TEST_KEY_TABLET in config_file_contents['TestBeds'][0]['Name'].lower(): |
| tablet_id = get_device_serial_number('tablet', config_file_contents) |
| else: |
| tablet_id = None |
| auto_scene_switch = tablet_id is not None |
| |
| # Run through all scenes if user does not supply one and config file doesn't |
| # have specific scene name listed. |
| possible_scenes = _AUTO_SCENES if auto_scene_switch else _ALL_SCENES |
| if not scenes or '<scene-name>' in 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) |
| elif s in _GROUPED_SCENES: |
| expand_scene(s, temp_scenes) |
| else: |
| scene_str = 'scene' + s |
| if scene_str in possible_scenes: |
| temp_scenes.append(scene_str) |
| elif scene_str in _GROUPED_SCENES: |
| expand_scene(scene_str, temp_scenes) |
| else: |
| valid_scenes = False |
| raise ValueError(f'Unknown scene specified: {s}') |
| |
| # assign temp_scenes back to scenes and remove duplicates |
| scenes = sorted(set(temp_scenes), key=temp_scenes.index) |
| |
| logging.info('Running ITS on device: %s, camera: %s, scene: %s', |
| device_id, camera_id_combos, scenes) |
| |
| for camera_id in camera_id_combos: |
| test_params_content['camera'] = camera_id |
| results = {} |
| for s in _ALL_SCENES: |
| results[s] = {RESULT_KEY: RESULT_NOT_EXECUTED} |
| # A subdir in topdir will be created for each camera_id. All scene test |
| # output logs for each camera id will be stored in this subdir. |
| # This output log path is a mobly param : LogPath |
| cam_id_string = 'cam_id_%s' % ( |
| camera_id.replace(its_session_utils.SUB_CAMERA_SEPARATOR, '_')) |
| mobly_output_logs_path = os.path.join(topdir, cam_id_string) |
| os.mkdir(mobly_output_logs_path) |
| tot_pass = 0 |
| for s in scenes: |
| test_params_content['scene'] = s |
| results[s]['TEST_STATUS'] = [] |
| |
| # unit is millisecond for execution time record in CtsVerifier |
| scene_start_time = int(round(time.time() * 1000)) |
| scene_test_summary = f'Cam{camera_id} {s}' + '\n' |
| mobly_scene_output_logs_path = os.path.join(mobly_output_logs_path, s) |
| |
| if auto_scene_switch: |
| # Copy scene images onto the tablet |
| if s not in ['scene0', 'sensor_fusion']: |
| load_scenes_on_tablet(s, tablet_id) |
| else: |
| # Check manual scens for correctness |
| if s not in ['scene0']: |
| check_manual_scenes(device_id, camera_id, s, mobly_output_logs_path) |
| |
| scene_test_list = [] |
| config_file_contents['TestBeds'][0]['TestParams'] = test_params_content |
| # Add the MoblyParams to config.yml file with the path to store camera_id |
| # test results. This is a separate dict other than TestBeds. |
| mobly_params_dict = { |
| 'MoblyParams': { |
| 'LogPath': mobly_scene_output_logs_path |
| } |
| } |
| config_file_contents.update(mobly_params_dict) |
| logging.debug('Final config file contents: %s', config_file_contents) |
| new_yml_file_name = get_updated_yml_file(config_file_contents) |
| logging.info('Using %s as temporary config yml file', new_yml_file_name) |
| if camera_id.rfind(its_session_utils.SUB_CAMERA_SEPARATOR) == -1: |
| scene_dir = os.listdir( |
| os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests', s)) |
| for file_name in scene_dir: |
| if file_name.endswith('.py') and 'test' in file_name: |
| scene_test_list.append(file_name) |
| if _REPEATED_TESTS[s]: |
| for t in _REPEATED_TESTS[s]: |
| scene_test_list.append((os.path.join('tests', t[0], t[1] + '.py'))) |
| else: # sub-camera |
| if SUB_CAMERA_TESTS.get(s): |
| scene_test_list = [f'{test}.py' for test in SUB_CAMERA_TESTS[s]] |
| else: |
| scene_test_list = [] |
| scene_test_list.sort() |
| |
| # Run tests for scene |
| logging.info('Running tests for %s with camera %s', s, camera_id) |
| num_pass = 0 |
| num_skip = 0 |
| num_not_mandated_fail = 0 |
| num_fail = 0 |
| for test in scene_test_list: |
| # Handle repeated test |
| if 'tests/' in test: |
| cmd = [ |
| 'python3', |
| os.path.join(os.environ['CAMERA_ITS_TOP'], test), '-c', |
| '%s' % new_yml_file_name |
| ] |
| else: |
| cmd = [ |
| 'python3', |
| os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests', s, test), |
| '-c', |
| '%s' % new_yml_file_name |
| ] |
| # pylint: disable=subprocess-run-check |
| with open(MOBLY_TEST_SUMMARY_TXT_FILE, 'w') as fp: |
| output = subprocess.run(cmd, stdout=fp) |
| # pylint: enable=subprocess-run-check |
| |
| # Parse mobly info output logs to determine skip and not_yet_mandated |
| # tests. |
| with open(MOBLY_TEST_SUMMARY_TXT_FILE, 'r') as file: |
| test_code = output.returncode |
| test_failed = False |
| test_skipped = False |
| test_not_yet_mandated = False |
| line = file.read() |
| if 'Test skipped' in line: |
| return_string = 'SKIP ' |
| num_skip += 1 |
| test_skipped = True |
| |
| if 'Not yet mandated test' in line: |
| return_string = 'FAIL*' |
| num_not_mandated_fail += 1 |
| test_not_yet_mandated = True |
| |
| if test_code == 0 and not test_skipped: |
| return_string = 'PASS ' |
| num_pass += 1 |
| |
| if test_code == 1 and not test_not_yet_mandated: |
| return_string = 'FAIL ' |
| num_fail += 1 |
| test_failed = True |
| |
| os.remove(MOBLY_TEST_SUMMARY_TXT_FILE) |
| logging.info('%s %s/%s', return_string, s, test) |
| test_name = test.split('/')[-1].split('.')[0] |
| results[s]['TEST_STATUS'].append({'test':test_name,'status':return_string.strip()}) |
| msg_short = '%s %s' % (return_string, test) |
| scene_test_summary += msg_short + '\n' |
| |
| # unit is millisecond for execution time record in CtsVerifier |
| scene_end_time = int(round(time.time() * 1000)) |
| skip_string = '' |
| tot_tests = len(scene_test_list) |
| if num_skip > 0: |
| skipstr = f",{num_skip} test{'s' if num_skip > 1 else ''} skipped" |
| test_result = '%d / %d tests passed (%.1f%%)%s' % ( |
| num_pass + num_not_mandated_fail, len(scene_test_list) - num_skip, |
| 100.0 * float(num_pass + num_not_mandated_fail) / |
| (len(scene_test_list) - num_skip) |
| if len(scene_test_list) != num_skip else 100.0, skip_string) |
| logging.info(test_result) |
| if num_not_mandated_fail > 0: |
| logging.info('(*) %s not_yet_mandated tests failed', |
| num_not_mandated_fail) |
| |
| tot_pass += num_pass |
| logging.info('scene tests: %s, Total tests passed: %s', tot_tests, |
| tot_pass) |
| if tot_tests > 0: |
| logging.info('%s compatibility score: %.f/100\n', |
| s, 100 * num_pass / tot_tests) |
| scene_test_summary_path = os.path.join(mobly_scene_output_logs_path, |
| 'scene_test_summary.txt') |
| with open(scene_test_summary_path, 'w') as f: |
| f.write(scene_test_summary) |
| results[s][RESULT_KEY] = (RESULT_PASS if num_fail == 0 else RESULT_FAIL) |
| results[s][SUMMARY_KEY] = scene_test_summary_path |
| results[s][TIME_KEY_START] = scene_start_time |
| results[s][TIME_KEY_END] = scene_end_time |
| else: |
| logging.info('%s compatibility score: 0/100\n') |
| |
| # Delete temporary yml file after scene run. |
| new_yaml_file_path = os.path.join(YAML_FILE_DIR, new_yml_file_name) |
| os.remove(new_yaml_file_path) |
| |
| logging.info('Reporting ITS result to CtsVerifier') |
| report_result(device_id, camera_id, results) |
| logging.info('Test execution completed.') |
| |
| if __name__ == '__main__': |
| main() |