| # 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 glob |
| import json |
| import logging |
| import os |
| import os.path |
| import re |
| import subprocess |
| import sys |
| import tempfile |
| import time |
| import types |
| |
| import camera_properties_utils |
| import capture_request_utils |
| import image_processing_utils |
| import its_device_utils |
| import its_session_utils |
| import lighting_control_utils |
| import numpy as np |
| import yaml |
| |
| |
| 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' |
| TEST_KEY_GEN2 = 'gen2' |
| ACTIVITY_START_WAIT = 1.5 # seconds |
| MERGE_RESULTS_TIMEOUT = 3600 # seconds |
| |
| NUM_TRIES = 2 |
| RESULT_PASS = 'PASS' |
| RESULT_FAIL = 'FAIL' |
| RESULT_NOT_EXECUTED = 'NOT_EXECUTED' |
| RESULT_KEY = 'result' |
| METRICS_KEY = 'mpc_metrics' |
| PERFORMANCE_KEY = 'performance_metrics' |
| FEATURE_QUERY_KEY = 'feature_query_proto' |
| FEATURE_QUERY_PATH_KEY = 'feature_query_proto_path' |
| SUMMARY_KEY = 'summary' |
| RESULT_VALUES = (RESULT_PASS, RESULT_FAIL, RESULT_NOT_EXECUTED) |
| CTS_VERIFIER_PACKAGE_NAME = 'com.android.cts.verifier' |
| 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' |
| EXTRA_TABLET_NAME = 'camera.its.extra.TABLET_NAME' |
| TIME_KEY_START = 'start' |
| TIME_KEY_END = 'end' |
| VALID_CONTROLLERS = ('arduino', 'canakit') |
| _FRONT_CAMERA_ID = '1' |
| # recover replaced '_' in scene def |
| _INT_STR_DICT = types.MappingProxyType({'11': '1_1', '12': '1_2', '13': '1_3'}) |
| _MAIN_TESTBED = 0 |
| _PROPERTIES_TO_MATCH = ( |
| 'ro.product.model', 'ro.product.name', 'ro.build.display.id', 'ro.revision' |
| ) |
| |
| # Scenes that can be automated through tablet display |
| # 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 |
| _TABLET_SCENES = ( |
| 'scene0', 'scene1_1', 'scene1_2', 'scene1_3', 'scene2_a', 'scene2_b', |
| 'scene2_c', 'scene2_d', 'scene2_e', 'scene2_f', 'scene2_g', 'scene3', |
| 'scene4', 'scene6', 'scene7', 'scene8', 'scene9', |
| os.path.join('scene_extensions', 'scene_hdr'), |
| os.path.join('scene_extensions', 'scene_low_light'), |
| os.path.join('scene_tele', 'scene6_tele'), |
| os.path.join('scene_tele', 'scene7_tele'), |
| 'scene_video', |
| ) |
| |
| # Scenes that use the 'sensor_fusion' test rig |
| _MOTION_SCENES = ('sensor_fusion', 'feature_combination',) |
| |
| # Scenes that uses lighting control |
| _FLASH_SCENES = ('scene_flash',) |
| |
| # Scenes that uses checkerboard as chart |
| _CHECKERBOARD_SCENES = ('sensor_fusion', 'scene_flash', 'feature_combination',) |
| |
| # Scenes that have to be run manually regardless of configuration |
| _MANUAL_SCENES = ('scene5',) |
| |
| # Scene extensions |
| _EXTENSIONS_SCENES = (os.path.join('scene_extensions', 'scene_hdr'), |
| os.path.join('scene_extensions', 'scene_low_light'), |
| ) |
| # Scene tele |
| _TELE_SCENES = (os.path.join('scene_tele', 'scene6_tele'), |
| os.path.join('scene_tele', 'scene7_tele'), |
| ) |
| _GEN2_RIG_SCENES = ('scene_ip',) |
| # All possible scenes |
| _ALL_SCENES = (_TABLET_SCENES + _MANUAL_SCENES + _MOTION_SCENES + |
| _FLASH_SCENES + _GEN2_RIG_SCENES) |
| |
| # Scenes that are logically grouped and can be called as group |
| # scene6_tele is not grouped with scene6 because it requires extension rig |
| _GROUPED_SCENES = types.MappingProxyType({ |
| 'scene1': ('scene1_1', 'scene1_2', 'scene1_3'), |
| 'scene2': ('scene2_a', 'scene2_b', 'scene2_c', 'scene2_d', 'scene2_e', |
| 'scene2_f', 'scene2_g') |
| }) |
| |
| # Scene requirements for manual testing. |
| _SCENE_REQ = types.MappingProxyType({ |
| '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', |
| 'scene1_3': 'A grey card covering at least the middle 30% of the scene, ' |
| 'without a white border like scene1_1 or scene1_2', |
| 'scene2_a': 'The picture with 3 faces in tests/scene2_a/scene2_a.png', |
| 'scene2_b': 'The picture with 3 faces in tests/scene2_b/scene2_b.png', |
| 'scene2_c': 'The picture with 3 faces in tests/scene2_c/scene2_c.png', |
| 'scene2_d': 'The picture with 3 faces in tests/scene2_d/scene2_d.png', |
| 'scene2_e': 'The picture with 3 faces in tests/scene2_e/scene2_e.png', |
| 'scene2_f': 'The picture with 3 faces in tests/scene2_f/scene2_f.png', |
| 'scene2_g': 'The picture with 3 profile faces in ' |
| 'tests/scene2_g/scene2_g.png', |
| 'scene3': 'The ISO12233 chart', |
| 'scene4': 'A test chart of a circle covering at least the middle 50% of ' |
| 'the scene. See tests/scene4/scene4.png', |
| 'scene5': 'Capture images with a diffuser attached to the camera. See ' |
| 'source.android.com/docs/compatibility/cts/camera-its-tests#scene5/diffuser ' # pylint: disable line-too-long |
| 'for more details', |
| 'scene6': 'A grid of ArUco markers on a white background. ' |
| 'See tests/scene6/scene6.png', |
| 'scene7': 'The picture with 4 different colors, slanted edge and' |
| '4 ArUco markers. See tests/scene7/scene7.png', |
| 'scene8': 'The picture with 4 faces in 4 different colors overlay.' |
| 'See tests/scene8/scene8.png', |
| 'scene9': 'A scene with high entropy consisting of random size and colored ' |
| 'circles. See tests/scene9/scene9.png', |
| # Use os.path to avoid confusion on other platforms |
| os.path.join('scene_extensions', 'scene_hdr'): ( |
| 'A tablet displayed scene with a face on the left ' |
| 'and a low-contrast QR code on the right. ' |
| 'See tests/scene_extensions/scene_hdr/scene_hdr.png' |
| ), |
| os.path.join('scene_extensions', 'scene_low_light'): ( |
| 'A tablet displayed scene with a grid of squares of varying ' |
| 'brightness. See ' |
| 'tests/scene_extensions/scene_low_light/scene_low_light.png' |
| ), |
| os.path.join('scene_tele', 'scene6_tele'): ( |
| 'Identical to scene6, but for tele cameras. ' |
| 'See tests/scene_tele/scene6_tele/scene6_tele.png' |
| ), |
| os.path.join('scene_tele', 'scene7_tele'): ( |
| 'Identical to scene7, but for tele cameras. ' |
| 'See tests/scene_tele/scene7_tele/scene7_tele.png' |
| ), |
| '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.', |
| 'feature_combination': 'The same scene as sensor_fusion, ' |
| 'separated for easier testing.', |
| 'scene_flash': 'A checkerboard pattern chart with lights off.', |
| 'scene_video': 'A tablet displayed scene with a series of circles moving ' |
| 'at different simulated frame rates. ' |
| 'See tests/scene_video/scene_video.mp4', |
| 'scene_ip': 'A chart with features such as QR code, color checker chart, ' |
| 'dead leaf patch, dynamic range chart used to analyze metrics ' |
| 'such as brightness, sharpness, color accuracy.', |
| }) |
| |
| # Made mutable to allow for test augmentation based on first API level |
| SUB_CAMERA_TESTS = { |
| 'scene0': ( |
| 'test_jitter', |
| 'test_metadata', |
| 'test_request_capture_match', |
| 'test_sensor_events', |
| 'test_solid_color_test_pattern', |
| 'test_unified_timestamps', |
| ), |
| 'scene1_1': ( |
| 'test_burst_capture', |
| 'test_burst_sameness_manual', |
| 'test_exposure_x_iso', |
| 'test_linearity', |
| ), |
| 'scene1_2': ( |
| 'test_raw_exposure', |
| ), |
| 'scene1_3': ( |
| 'test_dng_noise_model', |
| 'test_raw_sensitivity', |
| 'test_yuv_plus_raw', |
| ), |
| 'scene2_a': ( |
| 'test_num_faces', |
| ), |
| 'scene4': ( |
| 'test_aspect_ratio_and_crop', |
| ), |
| 'scene_tele/scene6_tele': ( |
| 'test_zoom_tele', |
| 'test_preview_zoom_tele', |
| ), |
| 'scene_tele/scene7_tele': ( |
| 'test_multi_camera_switch_tele', |
| ), |
| 'scene_video': ( |
| 'test_preview_frame_drop', |
| ), |
| 'sensor_fusion': ( |
| 'test_sensor_fusion', |
| ), |
| } |
| |
| _LIGHTING_CONTROL_TESTS = ( |
| 'test_auto_flash.py', |
| 'test_preview_min_frame_rate.py', |
| 'test_led_snapshot.py', |
| 'test_night_extension.py', |
| 'test_low_light_boost_extension.py', |
| 'test_hdr_extension.py', |
| ) |
| |
| _EXTENSION_NAMES = ( |
| 'hdr', |
| 'low_light', |
| ) |
| |
| _DST_SCENE_DIR = '/sdcard/Download/' |
| _DST_ROOT_DIR = '/sdcard/' |
| _SUB_CAMERA_LEVELS = 2 |
| MOBLY_TEST_SUMMARY_TXT_FILE = 'test_mobly_summary.txt' |
| |
| |
| def report_result(device_id, camera_id, tablet_name, 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. |
| tablet_name: The tablet name to identify model and build. |
| 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}' |
| its_device_utils.start_its_test_activity(device_id) |
| 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' |
| its_device_utils.run( |
| f'{adb} push {results[scene][SUMMARY_KEY]} {device_summary_path}') |
| results[scene][SUMMARY_KEY] = device_summary_path |
| if (FEATURE_QUERY_KEY in results[scene] and |
| FEATURE_QUERY_PATH_KEY in results[scene]): |
| for (proto_path, proto_file) in zip( |
| results[scene][FEATURE_QUERY_PATH_KEY], |
| results[scene][FEATURE_QUERY_KEY]): |
| host_path = os.path.join(proto_path, proto_file) |
| its_device_utils.run( |
| f'{adb} push {host_path} {_DST_ROOT_DIR}') |
| |
| 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_TABLET_NAME} {tablet_name} --es " |
| f"{EXTRA_RESULTS} \'{json_results}\'") |
| its_device_utils.run(cmd) |
| |
| |
| def write_result(testbed_index, device_id, camera_id, results): |
| """Writes results to a temporary file for merging. |
| |
| Args: |
| testbed_index: the index of a finished testbed. |
| device_id: the ID string of the device that created results. |
| camera_id: the ID string of the camera of the device. |
| results: a dictionary that contains all ITS scenes as key |
| and result/summary of current ITS run. |
| """ |
| result = {'device_id': device_id, 'results': results} |
| file_name = f'testbed_{testbed_index}_camera_{camera_id}.tmp' |
| with open(file_name, 'w') as f: |
| json.dump(result, f) |
| |
| |
| def parse_testbeds(completed_testbeds): |
| """Parses completed testbeds and yields device_id, camera_id, and results. |
| |
| Args: |
| completed_testbeds: an iterable of completed testbed indices. |
| Yields: |
| device_id: the device associated with the testbed. |
| camera_id: one of the camera_ids associated with the testbed. |
| results: the dictionary with scenes and result/summary of testbed's run. |
| """ |
| for i in completed_testbeds: |
| for file_name in glob.glob(f'testbed_{i}_camera_*.tmp'): |
| camera_id = file_name.split('camera_')[1].split('.tmp')[0] |
| device_id = '' |
| results = {} |
| with open(file_name, 'r') as f: |
| testbed_data = json.load(f) |
| device_id = testbed_data['device_id'] |
| results = testbed_data['results'] |
| if not device_id or not results: |
| raise ValueError(f'device_id or results for {file_name} not found.') |
| yield device_id, camera_id, results |
| |
| |
| def get_device_property(device_id, property_name): |
| """Get property of a given device. |
| |
| Args: |
| device_id: the ID string of a device. |
| property_name: the desired property string. |
| Returns: |
| The value of the property. |
| """ |
| property_cmd = f'adb -s {device_id} shell getprop {property_name}' |
| raw_output = subprocess.check_output( |
| property_cmd, stderr=subprocess.STDOUT, shell=True) |
| return str(raw_output.decode('utf-8')).strip() |
| |
| |
| def are_devices_similar(device_id_1, device_id_2): |
| """Checks if key dimensions are the same between devices. |
| |
| Args: |
| device_id_1: the ID string of the _MAIN_TESTBED device. |
| device_id_2: the ID string of another device. |
| Returns: |
| True if both devices share key dimensions. |
| """ |
| for property_to_match in _PROPERTIES_TO_MATCH: |
| property_value_1 = get_device_property(device_id_1, property_to_match) |
| property_value_2 = get_device_property(device_id_2, property_to_match) |
| if property_value_1 != property_value_2: |
| logging.error('%s does not match %s for %s', |
| property_value_1, property_value_2, property_to_match) |
| return False |
| return True |
| |
| |
| 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 |
| """ |
| hidden_physical_id = None |
| if its_session_utils.SUB_CAMERA_SEPARATOR in camera_id: |
| split_camera_ids = camera_id.split(its_session_utils.SUB_CAMERA_SEPARATOR) |
| if len(split_camera_ids) == _SUB_CAMERA_LEVELS: |
| camera_id = split_camera_ids[0] |
| hidden_physical_id = split_camera_ids[1] |
| |
| with its_session_utils.ItsSession( |
| device_id=device_id, |
| camera_id=camera_id, |
| hidden_physical_id=hidden_physical_id) as cam: |
| props = cam.get_camera_properties() |
| props = cam.override_with_hidden_physical_camera_props(props) |
| |
| 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 |
| if scene == 'scene5': |
| cam.do_3a(do_af=False, lock_ae=camera_properties_utils.ae_lock(props), |
| lock_awb=camera_properties_utils.awb_lock(props)) |
| else: |
| 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.replace("/", "_")}.jpg') |
| logging.info('Please check scene setup in %s', img_name) |
| image_processing_utils.write_image(img, img_name) |
| choice = input(f'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.safe_load(file) |
| 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 = str(device_dict.get('serial')) |
| if label == 'dut': |
| dut_device_id = str(device_dict.get('serial')) |
| if device == 'tablet': |
| return tablet_device_id |
| else: |
| return dut_device_id |
| |
| |
| 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) |
| file_descriptor, new_yaml_file = tempfile.mkstemp( |
| suffix='.yml', prefix='config_', dir=YAML_FILE_DIR) |
| os.close(file_descriptor) |
| 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 enable_external_storage(device_id): |
| """Override apk mode to allow write to external storage. |
| |
| Args: |
| device_id: Serial number of the device. |
| |
| """ |
| cmd = (f'adb -s {device_id} shell appops ' |
| 'set com.android.cts.verifier MANAGE_EXTERNAL_STORAGE allow') |
| its_device_utils.run(cmd) |
| |
| |
| def get_available_cameras(device_id, camera_id): |
| """Get available camera devices in the current state. |
| |
| Args: |
| device_id: Serial number of the device. |
| camera_id: Logical camera_id |
| |
| Returns: |
| List of all the available camera_ids. |
| """ |
| with its_session_utils.ItsSession( |
| device_id=device_id, |
| camera_id=camera_id) as cam: |
| props = cam.get_camera_properties() |
| props = cam.override_with_hidden_physical_camera_props(props) |
| unavailable_physical_cameras = cam.get_unavailable_physical_cameras( |
| camera_id) |
| unavailable_physical_ids = unavailable_physical_cameras[ |
| 'unavailablePhysicalCamerasArray'] |
| output = cam.get_camera_ids() |
| all_camera_ids = output['cameraIdArray'] |
| # Concat camera_id, physical camera_id and sub camera separator |
| unavailable_physical_ids = [f'{camera_id}.{s}' |
| for s in unavailable_physical_ids] |
| for i in unavailable_physical_ids: |
| if i in all_camera_ids: |
| all_camera_ids.remove(i) |
| logging.debug('available camera ids: %s', all_camera_ids) |
| return all_camera_ids |
| |
| |
| def get_unavailable_physical_cameras(device_id, camera_id): |
| """Get unavailable physical cameras in the current state. |
| |
| Args: |
| device_id: Serial number of the device. |
| camera_id: Logical camera device id |
| |
| Returns: |
| List of all the unavailable camera_ids. |
| """ |
| with its_session_utils.ItsSession( |
| device_id=device_id, |
| camera_id=camera_id) as cam: |
| unavailable_physical_cameras = cam.get_unavailable_physical_cameras( |
| camera_id) |
| unavailable_physical_ids = unavailable_physical_cameras[ |
| 'unavailablePhysicalCamerasArray'] |
| unavailable_physical_ids = [f'{camera_id}.{s}' |
| for s in unavailable_physical_ids] |
| logging.debug('Unavailable physical camera ids: %s', |
| unavailable_physical_ids) |
| return unavailable_physical_ids |
| |
| |
| def is_device_folded(device_id): |
| """Returns True if the foldable device is in folded state. |
| |
| Args: |
| device_id: Serial number of the foldable device. |
| """ |
| cmd = (f'adb -s {device_id} shell cmd device_state state') |
| result = subprocess.getoutput(cmd) |
| if 'CLOSE' in result: |
| return True |
| return False |
| |
| |
| def augment_sub_camera_tests(first_api_level): |
| """Adds certain tests to SUB_CAMERA_TESTS depending on first_api_level. |
| |
| Args: |
| first_api_level: First api level of the device. |
| """ |
| if (first_api_level >= its_session_utils.ANDROID15_API_LEVEL): |
| logging.debug('Augmenting sub camera tests') |
| SUB_CAMERA_TESTS['scene6'] = ('test_in_sensor_zoom',) |
| |
| |
| 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_') |
| try: |
| subprocess.call(['chmod', 'g+rx', topdir]) |
| except OSError as e: |
| logging.info(repr(e)) |
| |
| scenes = [] |
| camera_id_combos = [] |
| testbed_index = None |
| num_testbeds = None |
| |
| # Override camera, scenes and testbed with cmd line values if available |
| for s in list(sys.argv[1:]): |
| if 'scenes=' in s: |
| scenes = s.split('scenes=')[1].split(',') |
| elif 'camera=' in s: |
| camera_id_combos = s.split('=')[1].split(',') |
| elif 'testbed_index=' in s: |
| testbed_index = int(s.split('=')[1]) |
| elif 'num_testbeds=' in s: |
| num_testbeds = int(s.split('=')[1]) |
| else: |
| raise ValueError(f'Unknown argument {s}') |
| if testbed_index is None and num_testbeds is not None: |
| raise ValueError( |
| 'testbed_index must be specified if num_testbeds is specified.') |
| if (testbed_index is not None and num_testbeds is not None and |
| testbed_index >= num_testbeds): |
| raise ValueError('testbed_index must be less than num_testbeds. ' |
| 'testbed_index starts at 0.') |
| |
| # Prepend 'scene' if not specified at cmd line |
| for i, s in enumerate(scenes): |
| if (not s.startswith('scene') and |
| not s.startswith(('checkerboard', 'sensor_fusion', |
| 'flash', 'feature_combination', '<scene-name>'))): |
| scenes[i] = f'scene{s}' |
| if (s.startswith('flash') or s.startswith('extensions') |
| or s.startswith('ip')): |
| scenes[i] = f'scene_{s}' |
| # Handle scene_extensions |
| if any(s.startswith(extension) for extension in _EXTENSION_NAMES): |
| scenes[i] = f'scene_extensions/scene_{s}' |
| if (any(s.startswith('scene_' + extension) |
| for extension in _EXTENSION_NAMES)): |
| scenes[i] = f'scene_extensions/{s}' |
| # Handle scene_tele |
| if s == 'scene6_tele' or s == 'scene7_tele': |
| scenes[i] = f'scene_tele/{s}' |
| |
| # Read config file and extract relevant TestBed |
| config_file_contents = get_config_file_contents() |
| if testbed_index is None: |
| testbed_to_remove = [] |
| for i in config_file_contents['TestBeds']: |
| name = i['Name'].lower() |
| if scenes == ['scene_ip']: |
| if TEST_KEY_GEN2 not in name: |
| testbed_to_remove.append(i) |
| elif set(scenes).intersection( |
| set(_CHECKERBOARD_SCENES)) or scenes == ['checkerboard']: |
| if TEST_KEY_SENSOR_FUSION not in name: |
| testbed_to_remove.append(i) |
| else: |
| if TEST_KEY_SENSOR_FUSION in name or TEST_KEY_GEN2 in name: |
| testbed_to_remove.append(i) |
| for i in testbed_to_remove: |
| config_file_contents['TestBeds'].remove(i) |
| else: |
| config_file_contents = { |
| 'TestBeds': [config_file_contents['TestBeds'][testbed_index]] |
| } |
| |
| # 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 = str(test_params_content['scene']).split(',') |
| scenes = [_INT_STR_DICT.get(n, n) for n in scenes] # recover '1_1' & '1_2' |
| |
| device_id = get_device_serial_number('dut', config_file_contents) |
| # Enable external storage on DUT to send summary report to CtsVerifier.apk |
| enable_external_storage(device_id) |
| |
| # Add to SUB_CAMERA_TESTS depending on first_api_level |
| augment_sub_camera_tests(its_session_utils.get_first_api_level(device_id)) |
| |
| # Verify that CTS Verifier is installed |
| its_session_utils.check_apk_installed(device_id, CTS_VERIFIER_PACKAGE_NAME) |
| # Check whether the dut is foldable or not |
| testing_foldable_device = True if test_params_content[ |
| 'foldable_device'] == 'True' else False |
| available_camera_ids_to_test_foldable = [] |
| if testing_foldable_device: |
| logging.debug('Testing foldable device.') |
| # Check the state of foldable device. True if device is folded, |
| # false if the device is opened. |
| device_folded = is_device_folded(device_id) |
| # list of available camera_ids to be tested in device state |
| available_camera_ids_to_test_foldable = get_available_cameras( |
| device_id, _FRONT_CAMERA_ID) |
| |
| config_file_test_key = config_file_contents['TestBeds'][0]['Name'].lower() |
| logging.info('Saving %s output files to: %s', config_file_test_key, topdir) |
| if TEST_KEY_TABLET in config_file_test_key: |
| tablet_id = get_device_serial_number('tablet', config_file_contents) |
| tablet_name_cmd = f'adb -s {tablet_id} shell getprop ro.product.device' |
| raw_output = subprocess.check_output( |
| tablet_name_cmd, stderr=subprocess.STDOUT, shell=True) |
| tablet_name = str(raw_output.decode('utf-8')).strip() |
| logging.debug('Tablet name: %s', tablet_name) |
| brightness = test_params_content['brightness'] |
| its_session_utils.validate_tablet(tablet_name, brightness, tablet_id) |
| elif TEST_KEY_GEN2 in config_file_test_key: |
| tablet_id = None |
| tablet_name = 'ip_chart' |
| else: |
| tablet_id = None |
| tablet_name = 'sensor_fusion' |
| |
| testing_sensor_fusion_with_controller = False |
| if TEST_KEY_SENSOR_FUSION in config_file_test_key: |
| if test_params_content['rotator_cntl'].lower() in VALID_CONTROLLERS: |
| testing_sensor_fusion_with_controller = True |
| |
| testing_flash_with_controller = False |
| if (test_params_content.get('lighting_cntl', 'None').lower() == 'arduino' and |
| 'manual' not in config_file_test_key): |
| testing_flash_with_controller = True |
| |
| # Expand GROUPED_SCENES and remove any duplicates |
| scenes = [_GROUPED_SCENES[s] if s in _GROUPED_SCENES else s for s in scenes] |
| scenes = np.hstack(scenes).tolist() |
| scenes = sorted(set(scenes), key=scenes.index) |
| # List of scenes to be executed in folded state will have '_folded' |
| # prefix. This will help distinguish the test results from folded vs |
| # open device state for front camera_ids. |
| folded_device_scenes = [] |
| for scene in scenes: |
| folded_device_scenes.append(f'{scene}_folded') |
| |
| logging.info('Running ITS on device: %s, camera(s): %s, scene(s): %s', |
| device_id, camera_id_combos, scenes) |
| |
| # Determine if manual run |
| if tablet_id is not None and not set(scenes).intersection(_MANUAL_SCENES): |
| auto_scene_switch = True |
| else: |
| auto_scene_switch = False |
| logging.info('Manual, checkerboard scenes, scene5 or scene_ip testing.') |
| |
| folded_prompted = False |
| opened_prompted = False |
| for camera_id in camera_id_combos: |
| test_params_content['camera'] = camera_id |
| results = {} |
| unav_cameras = [] |
| # Get the list of unavailable cameras in current device state. |
| # These camera_ids should not be tested in current device state. |
| if testing_foldable_device: |
| unav_cameras = get_unavailable_physical_cameras( |
| device_id, _FRONT_CAMERA_ID) |
| |
| if testing_foldable_device: |
| device_state = 'folded' if device_folded else 'opened' |
| |
| testing_folded_front_camera = (testing_foldable_device and |
| device_folded and |
| _FRONT_CAMERA_ID in camera_id) |
| |
| # Raise an assertion error if there is any camera unavailable in |
| # current device state. Usually scenes with suffix 'folded' will |
| # be executed in folded state. |
| if (testing_foldable_device and |
| _FRONT_CAMERA_ID in camera_id and camera_id in unav_cameras): |
| raise AssertionError( |
| f'Camera {camera_id} is unavailable in device state {device_state}' |
| f' and cannot be tested with device {device_state}!') |
| |
| if (testing_folded_front_camera and camera_id not in unav_cameras |
| and not folded_prompted): |
| folded_prompted = True |
| input('\nYou are testing a foldable device in folded state. ' |
| 'Please make sure the device is folded and press <ENTER> ' |
| 'after positioning properly.\n') |
| |
| if (testing_foldable_device and |
| not device_folded and _FRONT_CAMERA_ID in camera_id and |
| camera_id not in unav_cameras and not opened_prompted): |
| opened_prompted = True |
| input('\nYou are testing a foldable device in opened state. ' |
| 'Please make sure the device is unfolded and press <ENTER> ' |
| 'after positioning properly.\n') |
| |
| # Run through all scenes if user does not supply one and config file doesn't |
| # have specific scene name listed. |
| if TEST_KEY_SENSOR_FUSION in config_file_test_key: |
| possible_scenes = _CHECKERBOARD_SCENES |
| elif its_session_utils.SUB_CAMERA_SEPARATOR in camera_id: |
| possible_scenes = list(SUB_CAMERA_TESTS.keys()) |
| if auto_scene_switch: |
| possible_scenes.remove('sensor_fusion') |
| if 'scene_tele' in scenes: |
| possible_scenes = _TELE_SCENES |
| else: |
| if 'scene_extensions' in scenes: |
| possible_scenes = _EXTENSIONS_SCENES |
| elif 'scene_tele' in scenes: |
| possible_scenes = _TELE_SCENES |
| elif 'scene_ip' in scenes: |
| possible_scenes = _GEN2_RIG_SCENES |
| else: |
| possible_scenes = _TABLET_SCENES if auto_scene_switch else _ALL_SCENES |
| if ('<scene-name>' in scenes or 'checkerboard' in scenes or |
| 'scene_extensions' in scenes or 'scene_tele' in scenes |
| or 'scene_ip' in scenes): |
| per_camera_scenes = possible_scenes |
| else: |
| # Validate user input scene names |
| per_camera_scenes = [] |
| for s in scenes: |
| if s in possible_scenes: |
| per_camera_scenes.append(s) |
| if not per_camera_scenes: |
| raise ValueError('No valid scene specified for this camera.') |
| |
| # Folded state scenes will have 'folded' suffix only for |
| # front cameras since rear cameras are common in both folded |
| # and unfolded state. |
| foldable_per_camera_scenes = [] |
| if testing_folded_front_camera: |
| if camera_id not in available_camera_ids_to_test_foldable: |
| raise AssertionError(f'camera {camera_id} is not available.') |
| for s in per_camera_scenes: |
| foldable_per_camera_scenes.append(f'{s}_folded') |
| |
| if foldable_per_camera_scenes: |
| per_camera_scenes = foldable_per_camera_scenes |
| |
| logging.info('camera: %s, scene(s): %s', camera_id, per_camera_scenes) |
| |
| if testing_folded_front_camera: |
| all_scenes = [f'{s}_folded' for s in _ALL_SCENES] |
| else: |
| all_scenes = _ALL_SCENES |
| |
| for s in all_scenes: |
| results[s] = {RESULT_KEY: RESULT_NOT_EXECUTED} |
| |
| # assert device folded testing scenes with suffix 'folded' |
| if testing_foldable_device and 'folded' in s: |
| if not device_folded: |
| raise AssertionError('Device should be folded during' |
| ' testing scenes with suffix "folded"') |
| |
| # 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 |
| camera_id_str = ( |
| camera_id.replace(its_session_utils.SUB_CAMERA_SEPARATOR, '_') |
| ) |
| mobly_output_logs_path = os.path.join(topdir, f'cam_id_{camera_id_str}') |
| os.mkdir(mobly_output_logs_path) |
| tot_pass = 0 |
| for s in per_camera_scenes: |
| results[s]['TEST_STATUS'] = [] |
| results[s][METRICS_KEY] = [] |
| results[s][PERFORMANCE_KEY] = [] |
| results[s][FEATURE_QUERY_KEY] = [] |
| results[s][FEATURE_QUERY_PATH_KEY] = [] |
| |
| # 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) |
| |
| # Since test directories do not have 'folded' in the name, we need |
| # to remove that suffix for the path of the scenes to be loaded |
| # on the tablets |
| testing_scene = s |
| if 'folded' in s: |
| testing_scene = s.split('_folded')[0] |
| test_params_content['scene'] = testing_scene |
| test_params_content['scene_with_suffix'] = s |
| |
| if auto_scene_switch: |
| # Copy scene images onto the tablet |
| if 'scene0' not in testing_scene: |
| its_session_utils.copy_scenes_to_tablet(testing_scene, tablet_id) |
| else: |
| # Check manual scenes for correctness |
| if ('scene0' not in testing_scene and |
| 'scene_ip' not in testing_scene and |
| not testing_sensor_fusion_with_controller): |
| check_manual_scenes(device_id, camera_id, testing_scene, |
| 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', testing_scene)) |
| for file_name in scene_dir: |
| if file_name.endswith('.py') and 'test' in file_name: |
| scene_test_list.append(file_name) |
| else: # sub-camera |
| if SUB_CAMERA_TESTS.get(testing_scene): |
| scene_test_list = [f'{test}.py' for test in SUB_CAMERA_TESTS[ |
| testing_scene]] |
| else: |
| scene_test_list = [] |
| scene_test_list.sort() |
| |
| # Run tests for scene |
| logging.info('Running tests for %s with camera %s', |
| testing_scene, 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', |
| f'{new_yml_file_name}' |
| ] |
| else: |
| cmd = [ |
| 'python3', |
| os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests', |
| testing_scene, test), |
| '-c', |
| f'{new_yml_file_name}' |
| ] |
| return_string = '' |
| for num_try in range(NUM_TRIES): |
| # Saves to mobly test summary file |
| # print only messages for manual lighting control testing |
| output = subprocess.Popen( |
| cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT |
| ) |
| with output.stdout, open( |
| os.path.join(topdir, MOBLY_TEST_SUMMARY_TXT_FILE), 'wb' |
| ) as file: |
| for line in iter(output.stdout.readline, b''): |
| out = line.decode('utf-8').strip() |
| if '<ENTER>' in out: print(out) |
| file.write(line) |
| output.wait() |
| |
| # Parse mobly logs to determine PASS/FAIL(*)/SKIP & socket FAILs |
| with open( |
| os.path.join(topdir, MOBLY_TEST_SUMMARY_TXT_FILE), 'r') as file: |
| test_code = output.returncode |
| test_skipped = False |
| test_not_yet_mandated = False |
| test_mpc_req = '' |
| perf_test_metrics = '' |
| hdr_mpc_req = '' |
| feature_query_proto = '' |
| content = file.read() |
| root_output_path = '' |
| |
| # Find media performance class logging |
| lines = content.splitlines() |
| for one_line in lines: |
| # regular expression pattern must match |
| # MPC12_CAMERA_LAUNCH_PATTERN or MPC12_JPEG_CAPTURE_PATTERN in |
| # ItsTestActivity.java. |
| mpc_string_match = re.search( |
| '^(1080p_jpeg_capture_time_ms:|camera_launch_time_ms:)', |
| one_line) |
| if mpc_string_match: |
| test_mpc_req = one_line |
| break |
| |
| for one_line in lines: |
| # regular expression pattern must match in ItsTestActivity.java. |
| gainmap_string_match = re.search('^has_gainmap:', one_line) |
| if gainmap_string_match: |
| hdr_mpc_req = one_line |
| break |
| |
| # Find root output path used to store feature combination query |
| # proto |
| for one_line in lines: |
| if 'root_output_path:' in one_line: |
| root_output_path = one_line.split(':')[1].strip() |
| # Find feature combination query proto |
| for one_line in lines: |
| # regular expression pattern must match in ItsTestActivity.java. |
| feature_comb_query_string_match = re.search( |
| '^feature_query_proto:(.*)', one_line |
| ) |
| if feature_comb_query_string_match: |
| feature_query_proto = feature_comb_query_string_match.group(1) |
| break |
| |
| # Find performance metrics logging |
| for one_line in lines: |
| # regular expression pattern must match in ItsTestActivity.java. |
| perf_metrics_string_match = re.search( |
| '^test.*:', |
| one_line) |
| if perf_metrics_string_match: |
| perf_test_metrics = one_line |
| # each test can add multiple metrics |
| results[s][PERFORMANCE_KEY].append(perf_test_metrics) |
| |
| if 'Test skipped' in content: |
| return_string = 'SKIP ' |
| num_skip += 1 |
| test_skipped = True |
| break |
| |
| if its_session_utils.NOT_YET_MANDATED_MESSAGE in content: |
| return_string = 'FAIL*' |
| num_not_mandated_fail += 1 |
| test_not_yet_mandated = True |
| break |
| |
| if test_code == 0 and not test_skipped: |
| return_string = 'PASS ' |
| num_pass += 1 |
| break |
| |
| if test_code != 0 and not test_not_yet_mandated: |
| return_string = 'FAIL ' |
| if 'Problem with socket' in content and num_try != NUM_TRIES-1: |
| logging.info('Retry %s/%s', s, test) |
| else: |
| num_fail += 1 |
| break |
| os.remove(os.path.join(topdir, MOBLY_TEST_SUMMARY_TXT_FILE)) |
| status_prefix = '' |
| if testbed_index is not None: |
| status_prefix = config_file_test_key + ':' |
| logging.info('%s%s %s/%s', status_prefix, return_string, s, test) |
| test_name = test.split('/')[-1].split('.')[0] |
| results[s]['TEST_STATUS'].append({ |
| 'test': test_name, |
| 'status': return_string.strip()}) |
| if test_mpc_req: |
| results[s][METRICS_KEY].append(test_mpc_req) |
| if hdr_mpc_req: |
| results[s][METRICS_KEY].append(hdr_mpc_req) |
| if feature_query_proto and root_output_path: |
| results[s][FEATURE_QUERY_KEY].append(feature_query_proto) |
| results[s][FEATURE_QUERY_PATH_KEY].append(root_output_path) |
| |
| msg_short = f'{return_string} {test}' |
| scene_test_summary += msg_short + '\n' |
| if (test in _LIGHTING_CONTROL_TESTS and |
| not testing_flash_with_controller): |
| print('Turn lights ON in rig and press <ENTER> to continue.') |
| |
| # 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) |
| tot_tests_run = tot_tests - num_skip |
| if tot_tests_run != 0: |
| tests_passed_ratio = (num_pass + num_not_mandated_fail) / tot_tests_run |
| else: |
| tests_passed_ratio = (num_pass + num_not_mandated_fail) / 100.0 |
| tests_passed_ratio_format = f'{(100 * tests_passed_ratio):.1f}%' |
| if num_skip > 0: |
| skip_string = f",{num_skip} test{'s' if num_skip > 1 else ''} skipped" |
| test_result = (f'{num_pass + num_not_mandated_fail} / {tot_tests_run} ' |
| f'tests passed ({tests_passed_ratio_format}){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) |
| # Log results per camera |
| if num_testbeds is None or testbed_index == _MAIN_TESTBED: |
| logging.info('Reporting camera %s ITS results to CtsVerifier', camera_id) |
| logging.info('ITS results to CtsVerifier: %s', results) |
| report_result(device_id, camera_id, tablet_name, results) |
| else: |
| write_result(testbed_index, device_id, camera_id, results) |
| |
| logging.info('Test execution completed.') |
| |
| # Power down tablet |
| if tablet_id: |
| cmd = f'adb -s {tablet_id} shell input keyevent KEYCODE_POWER' |
| subprocess.Popen(cmd.split()) |
| |
| # establish connection with lighting controller |
| lighting_cntl = test_params_content.get('lighting_cntl', 'None') |
| lighting_ch = test_params_content.get('lighting_ch', 'None') |
| arduino_serial_port = lighting_control_utils.lighting_control( |
| lighting_cntl, lighting_ch) |
| |
| # turn OFF lights |
| lighting_control_utils.set_lighting_state( |
| arduino_serial_port, lighting_ch, 'OFF') |
| |
| if num_testbeds is not None: |
| if testbed_index == _MAIN_TESTBED: |
| logging.info('Waiting for all testbeds to finish.') |
| start = time.time() |
| completed_testbeds = set() |
| while time.time() < start + MERGE_RESULTS_TIMEOUT: |
| for i in range(num_testbeds): |
| if os.path.isfile(f'testbed_{i}_completed.tmp'): |
| start = time.time() |
| completed_testbeds.add(i) |
| # Already reported _MAIN_TESTBED's results. |
| if len(completed_testbeds) == num_testbeds - 1: |
| logging.info('All testbeds completed, merging results.') |
| for parsed_id, parsed_camera, parsed_results in ( |
| parse_testbeds(completed_testbeds)): |
| logging.debug('Parsed id: %s, parsed cam: %s, parsed results: %s', |
| parsed_id, parsed_camera, parsed_results) |
| if not are_devices_similar(device_id, parsed_id): |
| logging.error('Device %s and device %s are not the same ' |
| 'model/type/build/revision.', |
| device_id, parsed_id) |
| return |
| report_result(device_id, parsed_camera, tablet_name, parsed_results) |
| for temp_file in glob.glob('testbed_*.tmp'): |
| os.remove(temp_file) |
| break |
| else: |
| logging.error('No testbeds finished in the last %d seconds, ' |
| 'but still expected data. ' |
| 'Completed testbed indices: %s, ' |
| 'expected number of testbeds: %d', |
| MERGE_RESULTS_TIMEOUT, list(completed_testbeds), |
| num_testbeds) |
| else: |
| with open(f'testbed_{testbed_index}_completed.tmp', 'w') as _: |
| pass |
| |
| if __name__ == '__main__': |
| main() |
