• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 json
16import logging
17import os
18import os.path
19import re
20import subprocess
21import sys
22import tempfile
23import time
24import yaml
25
26import capture_request_utils
27import camera_properties_utils
28import image_processing_utils
29import its_session_utils
30
31import numpy as np
32
33YAML_FILE_DIR = os.environ['CAMERA_ITS_TOP']
34CONFIG_FILE = os.path.join(YAML_FILE_DIR, 'config.yml')
35TEST_KEY_TABLET = 'tablet'
36TEST_KEY_SENSOR_FUSION = 'sensor_fusion'
37LOAD_SCENE_DELAY = 1  # seconds
38ACTIVITY_START_WAIT = 1.5  # seconds
39
40NUM_TRIES = 2
41RESULT_PASS = 'PASS'
42RESULT_FAIL = 'FAIL'
43RESULT_NOT_EXECUTED = 'NOT_EXECUTED'
44RESULT_KEY = 'result'
45METRICS_KEY = 'mpc_metrics'
46SUMMARY_KEY = 'summary'
47RESULT_VALUES = {RESULT_PASS, RESULT_FAIL, RESULT_NOT_EXECUTED}
48ITS_TEST_ACTIVITY = 'com.android.cts.verifier/.camera.its.ItsTestActivity'
49ACTION_ITS_RESULT = 'com.android.cts.verifier.camera.its.ACTION_ITS_RESULT'
50EXTRA_VERSION = 'camera.its.extra.VERSION'
51CURRENT_ITS_VERSION = '1.0'  # version number to sync with CtsVerifier
52EXTRA_CAMERA_ID = 'camera.its.extra.CAMERA_ID'
53EXTRA_RESULTS = 'camera.its.extra.RESULTS'
54TIME_KEY_START = 'start'
55TIME_KEY_END = 'end'
56VALID_CONTROLLERS = ('arduino', 'canakit')
57_INT_STR_DICT = {'11': '1_1', '12': '1_2'}  # recover replaced '_' in scene def
58_FRONT_CAMERA_ID = '1'
59
60# All possible scenes
61# Notes on scene names:
62#   scene*_1/2/... are same scene split to load balance run times for scenes
63#   scene*_a/b/... are similar scenes that share one or more tests
64_ALL_SCENES = [
65    'scene0', 'scene1_1', 'scene1_2', 'scene2_a', 'scene2_b', 'scene2_c',
66    'scene2_d', 'scene2_e', 'scene3', 'scene4', 'scene5', 'scene6',
67    'sensor_fusion'
68]
69
70# Scenes that can be automated through tablet display
71_AUTO_SCENES = [
72    'scene0', 'scene1_1', 'scene1_2', 'scene2_a', 'scene2_b', 'scene2_c',
73    'scene2_d', 'scene2_e', 'scene3', 'scene4', 'scene6'
74]
75
76# Scenes that are logically grouped and can be called as group
77_GROUPED_SCENES = {
78        'scene1': ['scene1_1', 'scene1_2'],
79        'scene2': ['scene2_a', 'scene2_b', 'scene2_c', 'scene2_d', 'scene2_e']
80}
81
82# Scenes that have to be run manually regardless of configuration
83_MANUAL_SCENES = ['scene5']
84
85# Scene requirements for manual testing.
86_SCENE_REQ = {
87    'scene0': None,
88    'scene1_1': 'A grey card covering at least the middle 30% of the scene',
89    'scene1_2': 'A grey card covering at least the middle 30% of the scene',
90    'scene2_a': 'The picture with 3 faces in tests/scene2_a/scene2_a.png',
91    'scene2_b': 'The picture with 3 faces in tests/scene2_b/scene2_b.png',
92    'scene2_c': 'The picture with 3 faces in tests/scene2_c/scene2_c.png',
93    'scene2_d': 'The picture with 3 faces in tests/scene2_d/scene2_d.png',
94    'scene2_e': 'The picture with 3 faces in tests/scene2_e/scene2_e.png',
95    'scene3': 'The ISO12233 chart',
96    'scene4': 'A test chart of a circle covering at least the middle 50% of '
97              'the scene. See tests/scene4/scene4.png',
98    'scene5': 'Capture images with a diffuser attached to the camera. '
99              'See CameraITS.pdf section 2.3.4 for more details',
100    'scene6': 'A grid of black circles on a white background. '
101              'See tests/scene6/scene6.png',
102    'sensor_fusion': 'A checkerboard pattern for phone to rotate in front of '
103                     'in tests/sensor_fusion/checkerboard.pdf\n'
104                     'See tests/sensor_fusion/SensorFusion.pdf for detailed '
105                     'instructions.\nNote that this test will be skipped '
106                     'on devices not supporting REALTIME camera timestamp.',
107}
108
109
110SUB_CAMERA_TESTS = {
111    'scene0': [
112        'test_burst_capture',
113        'test_jitter',
114        'test_metadata',
115        'test_read_write',
116        'test_sensor_events',
117        'test_solid_color_test_pattern',
118        'test_unified_timestamps',
119    ],
120    'scene1_1': [
121        'test_burst_sameness_manual',
122        'test_dng_noise_model',
123        'test_exposure',
124        'test_linearity',
125    ],
126    'scene1_2': [
127        'test_raw_exposure',
128        'test_raw_sensitivity',
129        'test_yuv_plus_raw',
130    ],
131    'scene2_a': [
132        'test_num_faces',
133    ],
134    'scene4': [
135        'test_aspect_ratio_and_crop',
136    ],
137    'sensor_fusion': [
138        'test_sensor_fusion',
139    ],
140}
141
142_LIGHTING_CONTROL_TESTS = [
143    'test_auto_flash.py'
144    ]
145
146_DST_SCENE_DIR = '/sdcard/Download/'
147MOBLY_TEST_SUMMARY_TXT_FILE = 'test_mobly_summary.txt'
148
149
150def run(cmd):
151  """Replaces os.system call, while hiding stdout+stderr messages."""
152  with open(os.devnull, 'wb') as devnull:
153    subprocess.check_call(cmd.split(), stdout=devnull, stderr=subprocess.STDOUT)
154
155
156def report_result(device_id, camera_id, results):
157  """Sends a pass/fail result to the device, via an intent.
158
159  Args:
160   device_id: The ID string of the device to report the results to.
161   camera_id: The ID string of the camera for which to report pass/fail.
162   results: a dictionary contains all ITS scenes as key and result/summary of
163            current ITS run. See test_report_result unit test for an example.
164  """
165  adb = f'adb -s {device_id}'
166
167  # Start ItsTestActivity to receive test results
168  cmd = f'{adb} shell am start {ITS_TEST_ACTIVITY} --activity-brought-to-front'
169  run(cmd)
170  time.sleep(ACTIVITY_START_WAIT)
171
172  # Validate/process results argument
173  for scene in results:
174    if RESULT_KEY not in results[scene]:
175      raise ValueError(f'ITS result not found for {scene}')
176    if results[scene][RESULT_KEY] not in RESULT_VALUES:
177      raise ValueError(f'Unknown ITS result for {scene}: {results[RESULT_KEY]}')
178    if SUMMARY_KEY in results[scene]:
179      device_summary_path = f'/sdcard/its_camera{camera_id}_{scene}.txt'
180      run('%s push %s %s' %
181          (adb, results[scene][SUMMARY_KEY], device_summary_path))
182      results[scene][SUMMARY_KEY] = device_summary_path
183
184  json_results = json.dumps(results)
185  cmd = (f"{adb} shell am broadcast -a {ACTION_ITS_RESULT} --es {EXTRA_VERSION}"
186         f" {CURRENT_ITS_VERSION} --es {EXTRA_CAMERA_ID} {camera_id} --es "
187         f"{EXTRA_RESULTS} \'{json_results}\'")
188  if len(cmd) > 8000:
189    logging.info('ITS command string might be too long! len:%s', len(cmd))
190  run(cmd)
191
192
193def load_scenes_on_tablet(scene, tablet_id):
194  """Copies scenes onto the tablet before running the tests.
195
196  Args:
197    scene: Name of the scene to copy image files.
198    tablet_id: adb id of tablet
199  """
200  logging.info('Copying files to tablet: %s', tablet_id)
201  scene_dir = os.listdir(
202      os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests', scene))
203  for file_name in scene_dir:
204    if file_name.endswith('.png'):
205      src_scene_file = os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests',
206                                    scene, file_name)
207      cmd = f'adb -s {tablet_id} push {src_scene_file} {_DST_SCENE_DIR}'
208      subprocess.Popen(cmd.split())
209  time.sleep(LOAD_SCENE_DELAY)
210  logging.info('Finished copying files to tablet.')
211
212
213def check_manual_scenes(device_id, camera_id, scene, out_path):
214  """Halt run to change scenes.
215
216  Args:
217    device_id: id of device
218    camera_id: id of camera
219    scene: Name of the scene to copy image files.
220    out_path: output file location
221  """
222  with its_session_utils.ItsSession(
223      device_id=device_id,
224      camera_id=camera_id) as cam:
225    props = cam.get_camera_properties()
226    props = cam.override_with_hidden_physical_camera_props(props)
227
228    while True:
229      input(f'\n Press <ENTER> after positioning camera {camera_id} with '
230            f'{scene}.\n The scene setup should be: \n  {_SCENE_REQ[scene]}\n')
231      # Converge 3A prior to capture
232      if scene == 'scene5':
233        cam.do_3a(do_af=False, lock_ae=camera_properties_utils.ae_lock(props),
234                  lock_awb=camera_properties_utils.awb_lock(props))
235      else:
236        cam.do_3a()
237      req, fmt = capture_request_utils.get_fastest_auto_capture_settings(props)
238      logging.info('Capturing an image to check the test scene')
239      cap = cam.do_capture(req, fmt)
240      img = image_processing_utils.convert_capture_to_rgb_image(cap)
241      img_name = os.path.join(out_path, f'test_{scene}.jpg')
242      logging.info('Please check scene setup in %s', img_name)
243      image_processing_utils.write_image(img, img_name)
244      choice = input(f'Is the image okay for ITS {scene}? (Y/N)').lower()
245      if choice == 'y':
246        break
247
248
249def get_config_file_contents():
250  """Read the config file contents from a YML file.
251
252  Args:
253    None
254
255  Returns:
256    config_file_contents: a dict read from config.yml
257  """
258  with open(CONFIG_FILE) as file:
259    config_file_contents = yaml.safe_load(file)
260  return config_file_contents
261
262
263def get_test_params(config_file_contents):
264  """Reads the config file parameters.
265
266  Args:
267    config_file_contents: dict read from config.yml file
268
269  Returns:
270    dict of test parameters
271  """
272  test_params = None
273  for _, j in config_file_contents.items():
274    for datadict in j:
275      test_params = datadict.get('TestParams')
276  return test_params
277
278
279def get_device_serial_number(device, config_file_contents):
280  """Returns the serial number of the device with label from the config file.
281
282  The config file contains TestBeds dictionary which contains Controllers and
283  Android Device dicts.The two devices used by the test per box are listed
284  here labels dut and tablet. Parse through the nested TestBeds dict to get
285  the Android device details.
286
287  Args:
288    device: String device label as specified in config file.dut/tablet
289    config_file_contents: dict read from config.yml file
290  """
291
292  for _, j in config_file_contents.items():
293    for datadict in j:
294      android_device_contents = datadict.get('Controllers')
295      for device_dict in android_device_contents.get('AndroidDevice'):
296        for _, label in device_dict.items():
297          if label == 'tablet':
298            tablet_device_id = str(device_dict.get('serial'))
299          if label == 'dut':
300            dut_device_id = str(device_dict.get('serial'))
301  if device == 'tablet':
302    return tablet_device_id
303  else:
304    return dut_device_id
305
306
307def get_updated_yml_file(yml_file_contents):
308  """Create a new yml file and write the testbed contents in it.
309
310  This testbed file is per box and contains all the parameters and
311  device id used by the mobly tests.
312
313  Args:
314   yml_file_contents: Data to write in yml file.
315
316  Returns:
317    Updated yml file contents.
318  """
319  os.chmod(YAML_FILE_DIR, 0o755)
320  file_descriptor, new_yaml_file = tempfile.mkstemp(
321      suffix='.yml', prefix='config_', dir=YAML_FILE_DIR)
322  os.close(file_descriptor)
323  with open(new_yaml_file, 'w') as f:
324    yaml.dump(yml_file_contents, stream=f, default_flow_style=False)
325  new_yaml_file_name = os.path.basename(new_yaml_file)
326  return new_yaml_file_name
327
328
329def enable_external_storage(device_id):
330  """Override apk mode to allow write to external storage.
331
332  Args:
333    device_id: Serial number of the device.
334
335  """
336  cmd = (f'adb -s {device_id} shell appops '
337         'set com.android.cts.verifier MANAGE_EXTERNAL_STORAGE allow')
338  run(cmd)
339
340
341def get_available_cameras(device_id, camera_id):
342  """Get available camera devices in the current state.
343
344  Args:
345    device_id: Serial number of the device.
346    camera_id: Logical camera_id
347
348  Returns:
349    List of all the available camera_ids.
350  """
351  with its_session_utils.ItsSession(
352      device_id=device_id,
353      camera_id=camera_id) as cam:
354    props = cam.get_camera_properties()
355    props = cam.override_with_hidden_physical_camera_props(props)
356    unavailable_physical_cameras = cam.get_unavailable_physical_cameras(
357        camera_id)
358    unavailable_physical_ids = unavailable_physical_cameras[
359        'unavailablePhysicalCamerasArray']
360    output = cam.get_camera_ids()
361    all_camera_ids = output['cameraIdArray']
362    # Concat camera_id, physical camera_id and sub camera separator
363    unavailable_physical_ids = [f'{camera_id}.{s}'
364                                for s in unavailable_physical_ids]
365    for i in unavailable_physical_ids:
366      if i in all_camera_ids:
367        all_camera_ids.remove(i)
368    logging.debug('available camera ids: %s', all_camera_ids)
369  return all_camera_ids
370
371
372def get_unavailable_physical_cameras(device_id, camera_id):
373  """Get unavailable physical cameras in the current state.
374
375  Args:
376    device_id: Serial number of the device.
377    camera_id: Logical camera device id
378
379  Returns:
380    List of all the unavailable camera_ids.
381  """
382  with its_session_utils.ItsSession(
383      device_id=device_id,
384      camera_id=camera_id) as cam:
385    unavailable_physical_cameras = cam.get_unavailable_physical_cameras(
386        camera_id)
387    unavailable_physical_ids = unavailable_physical_cameras[
388        'unavailablePhysicalCamerasArray']
389    unavailable_physical_ids = [f'{camera_id}.{s}'
390                                for s in unavailable_physical_ids]
391    logging.debug('Unavailable physical camera ids: %s',
392                  unavailable_physical_ids)
393  return unavailable_physical_ids
394
395
396def is_device_folded(device_id):
397  """Returns True if the foldable device is in folded state.
398
399  Args:
400    device_id: Serial number of the foldable device.
401  """
402  cmd = (f'adb -s {device_id} shell cmd device_state state')
403  result = subprocess.getoutput(cmd)
404  if 'CLOSED' in result:
405    return True
406  return False
407
408
409def main():
410  """Run all the Camera ITS automated tests.
411
412    Script should be run from the top-level CameraITS directory.
413
414    Command line arguments:
415        camera:  the camera(s) to be tested. Use comma to separate multiple
416                 camera Ids. Ex: "camera=0,1" or "camera=1"
417        scenes:  the test scene(s) to be executed. Use comma to separate
418                 multiple scenes. Ex: "scenes=scene0,scene1_1" or
419                 "scenes=0,1_1,sensor_fusion" (sceneX can be abbreviated by X
420                 where X is scene name minus 'scene')
421  """
422  logging.basicConfig(level=logging.INFO)
423  # Make output directories to hold the generated files.
424  topdir = tempfile.mkdtemp(prefix='CameraITS_')
425  try:
426    subprocess.call(['chmod', 'g+rx', topdir])
427  except OSError as e:
428    logging.info(repr(e))
429  logging.info('Saving output files to: %s', topdir)
430
431  scenes = []
432  camera_id_combos = []
433  # Override camera & scenes with cmd line values if available
434  for s in list(sys.argv[1:]):
435    if 'scenes=' in s:
436      scenes = s.split('=')[1].split(',')
437    elif 'camera=' in s:
438      camera_id_combos = s.split('=')[1].split(',')
439
440  # Read config file and extract relevant TestBed
441  config_file_contents = get_config_file_contents()
442  for i in config_file_contents['TestBeds']:
443    if scenes == ['sensor_fusion']:
444      if TEST_KEY_SENSOR_FUSION not in i['Name'].lower():
445        config_file_contents['TestBeds'].remove(i)
446    else:
447      if TEST_KEY_SENSOR_FUSION in i['Name'].lower():
448        config_file_contents['TestBeds'].remove(i)
449
450  # Get test parameters from config file
451  test_params_content = get_test_params(config_file_contents)
452  if not camera_id_combos:
453    camera_id_combos = str(test_params_content['camera']).split(',')
454  if not scenes:
455    scenes = str(test_params_content['scene']).split(',')
456    scenes = [_INT_STR_DICT.get(n, n) for n in scenes]  # recover '1_1' & '1_2'
457
458  device_id = get_device_serial_number('dut', config_file_contents)
459  # Enable external storage on DUT to send summary report to CtsVerifier.apk
460  enable_external_storage(device_id)
461
462  # Check whether the dut is foldable or not
463  testing_foldable_device = True if test_params_content[
464      'foldable_device'] == 'True' else False
465  available_camera_ids_to_test_foldable = []
466  if testing_foldable_device:
467    logging.debug('Testing foldable device.')
468    # Check the state of foldable device. True if device is folded,
469    # false if the device is opened.
470    device_folded = is_device_folded(device_id)
471    # list of available camera_ids to be tested in device state
472    available_camera_ids_to_test_foldable = get_available_cameras(
473        device_id, _FRONT_CAMERA_ID)
474
475  config_file_test_key = config_file_contents['TestBeds'][0]['Name'].lower()
476  if TEST_KEY_TABLET in config_file_test_key:
477    tablet_id = get_device_serial_number('tablet', config_file_contents)
478    tablet_name_cmd = f'adb -s {tablet_id} shell getprop ro.build.product'
479    raw_output = subprocess.check_output(
480        tablet_name_cmd, stderr=subprocess.STDOUT, shell=True)
481    tablet_name = str(raw_output.decode('utf-8')).strip()
482    logging.debug('Tablet name: %s', tablet_name)
483    brightness = test_params_content['brightness']
484    its_session_utils.validate_tablet_brightness(tablet_name, brightness)
485  else:
486    tablet_id = None
487
488  testing_sensor_fusion_with_controller = False
489  if TEST_KEY_SENSOR_FUSION in config_file_test_key:
490    if test_params_content['rotator_cntl'].lower() in VALID_CONTROLLERS:
491      testing_sensor_fusion_with_controller = True
492
493  testing_flash_with_controller = False
494  if (TEST_KEY_TABLET in config_file_test_key or
495      'manual' in config_file_test_key):
496    if test_params_content['lighting_cntl'].lower() == 'arduino':
497      testing_flash_with_controller = True
498
499  # Prepend 'scene' if not specified at cmd line
500  for i, s in enumerate(scenes):
501    if (not s.startswith('scene') and
502        not s.startswith(('sensor_fusion', '<scene-name>'))):
503      scenes[i] = f'scene{s}'
504
505  # Expand GROUPED_SCENES and remove any duplicates
506  scenes = [_GROUPED_SCENES[s] if s in _GROUPED_SCENES else s for s in scenes]
507  scenes = np.hstack(scenes).tolist()
508  scenes = sorted(set(scenes), key=scenes.index)
509  # List of scenes to be executed in folded state will have '_folded'
510  # prefix. This will help distinguish the test results from folded vs
511  # open device state for front camera_ids.
512  folded_device_scenes = []
513  for scene in scenes:
514    folded_device_scenes.append(f'{scene}_folded')
515
516  logging.info('Running ITS on device: %s, camera(s): %s, scene(s): %s',
517               device_id, camera_id_combos, scenes)
518
519  # Determine if manual run
520  if tablet_id is not None and not set(scenes).intersection(_MANUAL_SCENES):
521    auto_scene_switch = True
522  else:
523    auto_scene_switch = False
524    logging.info('No tablet: manual, sensor_fusion, or scene5 testing.')
525
526  folded_prompted = False
527  opened_prompted = False
528  for camera_id in camera_id_combos:
529    test_params_content['camera'] = camera_id
530    results = {}
531    unav_cameras = []
532    # Get the list of unavailable cameras in current device state.
533    # These camera_ids should not be tested in current device state.
534    if testing_foldable_device:
535      unav_cameras = get_unavailable_physical_cameras(
536          device_id, _FRONT_CAMERA_ID)
537
538    if testing_foldable_device:
539      device_state = 'folded' if device_folded else 'opened'
540
541    testing_folded_front_camera = (testing_foldable_device and
542                                   device_folded and
543                                   _FRONT_CAMERA_ID in camera_id)
544
545    # Raise an assertion error if there is any camera unavailable in
546    # current device state. Usually scenes with suffix 'folded' will
547    # be executed in folded state.
548    if (testing_foldable_device and
549        _FRONT_CAMERA_ID in camera_id and camera_id in unav_cameras):
550      raise AssertionError(
551          f'Camera {camera_id} is unavailable in device state {device_state}'
552          f' and cannot be tested with device {device_state}!')
553
554    if (testing_folded_front_camera and camera_id not in unav_cameras
555        and not folded_prompted):
556      folded_prompted = True
557      input('\nYou are testing a foldable device in folded state.'
558            'Please make sure the device is folded and press <ENTER>'
559            'after positioning properly.\n')
560
561    if (testing_foldable_device and
562        not device_folded and _FRONT_CAMERA_ID in camera_id and
563        camera_id not in unav_cameras and not opened_prompted):
564      opened_prompted = True
565      input('\nYou are testing a foldable device in opened state.'
566            'Please make sure the device is unfolded and press <ENTER>'
567            'after positioning properly.\n')
568
569    # Run through all scenes if user does not supply one and config file doesn't
570    # have specific scene name listed.
571    if its_session_utils.SUB_CAMERA_SEPARATOR in camera_id:
572      possible_scenes = list(SUB_CAMERA_TESTS.keys())
573      if auto_scene_switch:
574        possible_scenes.remove('sensor_fusion')
575    else:
576      possible_scenes = _AUTO_SCENES if auto_scene_switch else _ALL_SCENES
577
578    if '<scene-name>' in scenes:
579      per_camera_scenes = possible_scenes
580    else:
581      # Validate user input scene names
582      per_camera_scenes = []
583      for s in scenes:
584        if s in possible_scenes:
585          per_camera_scenes.append(s)
586      if not per_camera_scenes:
587        raise ValueError('No valid scene specified for this camera.')
588
589    # Folded state scenes will have 'folded' suffix only for
590    # front cameras since rear cameras are common in both folded
591    # and unfolded state.
592    foldable_per_camera_scenes = []
593    if testing_folded_front_camera:
594      if camera_id not in available_camera_ids_to_test_foldable:
595        raise AssertionError(f'camera {camera_id} is not available.')
596      for s in per_camera_scenes:
597        foldable_per_camera_scenes.append(f'{s}_folded')
598
599    if foldable_per_camera_scenes:
600      per_camera_scenes = foldable_per_camera_scenes
601
602    logging.info('camera: %s, scene(s): %s', camera_id, per_camera_scenes)
603
604    if testing_folded_front_camera:
605      all_scenes = [f'{s}_folded' for s in _ALL_SCENES]
606    else:
607      all_scenes = _ALL_SCENES
608
609    for s in all_scenes:
610      results[s] = {RESULT_KEY: RESULT_NOT_EXECUTED}
611
612      # assert device folded testing scenes with suffix 'folded'
613      if testing_foldable_device and 'folded' in s:
614        if not device_folded:
615          raise AssertionError('Device should be folded during'
616                               ' testing scenes with suffix "folded"')
617
618    # A subdir in topdir will be created for each camera_id. All scene test
619    # output logs for each camera id will be stored in this subdir.
620    # This output log path is a mobly param : LogPath
621    cam_id_string = 'cam_id_%s' % (
622        camera_id.replace(its_session_utils.SUB_CAMERA_SEPARATOR, '_'))
623    mobly_output_logs_path = os.path.join(topdir, cam_id_string)
624    os.mkdir(mobly_output_logs_path)
625    tot_pass = 0
626    for s in per_camera_scenes:
627      results[s]['TEST_STATUS'] = []
628      results[s][METRICS_KEY] = []
629
630      # unit is millisecond for execution time record in CtsVerifier
631      scene_start_time = int(round(time.time() * 1000))
632      scene_test_summary = f'Cam{camera_id} {s}' + '\n'
633      mobly_scene_output_logs_path = os.path.join(mobly_output_logs_path, s)
634
635      # Since test directories do not have 'folded' in the name, we need
636      # to remove that suffix for the path of the scenes to be loaded
637      # on the tablets
638      testing_scene = s
639      if 'folded' in s:
640        testing_scene = s.split('_folded')[0]
641      test_params_content['scene'] = testing_scene
642      if auto_scene_switch:
643        # Copy scene images onto the tablet
644        if 'scene0' not in testing_scene:
645          load_scenes_on_tablet(testing_scene, tablet_id)
646      else:
647        # Check manual scenes for correctness
648        if 'scene0' not in testing_scene and not testing_sensor_fusion_with_controller:
649          check_manual_scenes(device_id, camera_id, testing_scene,
650                              mobly_output_logs_path)
651
652      scene_test_list = []
653      config_file_contents['TestBeds'][0]['TestParams'] = test_params_content
654      # Add the MoblyParams to config.yml file with the path to store camera_id
655      # test results. This is a separate dict other than TestBeds.
656      mobly_params_dict = {
657          'MoblyParams': {
658              'LogPath': mobly_scene_output_logs_path
659          }
660      }
661      config_file_contents.update(mobly_params_dict)
662      logging.debug('Final config file contents: %s', config_file_contents)
663      new_yml_file_name = get_updated_yml_file(config_file_contents)
664      logging.info('Using %s as temporary config yml file', new_yml_file_name)
665      if camera_id.rfind(its_session_utils.SUB_CAMERA_SEPARATOR) == -1:
666        scene_dir = os.listdir(
667            os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests', testing_scene))
668        for file_name in scene_dir:
669          if file_name.endswith('.py') and 'test' in file_name:
670            scene_test_list.append(file_name)
671      else:  # sub-camera
672        if SUB_CAMERA_TESTS.get(testing_scene):
673          scene_test_list = [f'{test}.py' for test in SUB_CAMERA_TESTS[
674              testing_scene]]
675        else:
676          scene_test_list = []
677      scene_test_list.sort()
678
679      # Run tests for scene
680      logging.info('Running tests for %s with camera %s',
681                   testing_scene, camera_id)
682      num_pass = 0
683      num_skip = 0
684      num_not_mandated_fail = 0
685      num_fail = 0
686      for test in scene_test_list:
687        # Handle repeated test
688        if 'tests/' in test:
689          cmd = [
690              'python3',
691              os.path.join(os.environ['CAMERA_ITS_TOP'], test), '-c',
692              '%s' % new_yml_file_name
693          ]
694        else:
695          cmd = [
696              'python3',
697              os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests',
698                           testing_scene, test),
699              '-c',
700              '%s' % new_yml_file_name
701          ]
702        for num_try in range(NUM_TRIES):
703          # Handle manual lighting control redirected stdout in test
704          if (test in _LIGHTING_CONTROL_TESTS and
705              not testing_flash_with_controller):
706            print('Turn lights OFF in rig and press <ENTER> to continue.')
707
708          # pylint: disable=subprocess-run-check
709          with open(MOBLY_TEST_SUMMARY_TXT_FILE, 'w') as fp:
710            output = subprocess.run(cmd, stdout=fp)
711          # pylint: enable=subprocess-run-check
712
713          # Parse mobly logs to determine SKIP, NOT_YET_MANDATED, and
714          # socket FAILs.
715          with open(MOBLY_TEST_SUMMARY_TXT_FILE, 'r') as file:
716            test_code = output.returncode
717            test_skipped = False
718            test_not_yet_mandated = False
719            test_mpc_req = ''
720            content = file.read()
721
722            # Find media performance class logging
723            lines = content.splitlines()
724            for one_line in lines:
725              # regular expression pattern must match
726              # MPC12_CAMERA_LAUNCH_PATTERN or MPC12_JPEG_CAPTURE_PATTERN in
727              # ItsTestActivity.java.
728              mpc_string_match = re.search(
729                  '^(1080p_jpeg_capture_time_ms:|camera_launch_time_ms:)',
730                  one_line)
731              if mpc_string_match:
732                test_mpc_req = one_line
733                break
734
735            if 'Test skipped' in content:
736              return_string = 'SKIP '
737              num_skip += 1
738              test_skipped = True
739              break
740
741            if 'Not yet mandated test' in content:
742              return_string = 'FAIL*'
743              num_not_mandated_fail += 1
744              test_not_yet_mandated = True
745              break
746
747            if test_code == 0 and not test_skipped:
748              return_string = 'PASS '
749              num_pass += 1
750              break
751
752            if test_code == 1 and not test_not_yet_mandated:
753              return_string = 'FAIL '
754              if 'Problem with socket' in content and num_try != NUM_TRIES-1:
755                logging.info('Retry %s/%s', s, test)
756              else:
757                num_fail += 1
758                break
759            os.remove(MOBLY_TEST_SUMMARY_TXT_FILE)
760        logging.info('%s %s/%s', return_string, s, test)
761        test_name = test.split('/')[-1].split('.')[0]
762        results[s]['TEST_STATUS'].append({'test': test_name,
763                                          'status': return_string.strip()})
764        if test_mpc_req:
765          results[s][METRICS_KEY].append(test_mpc_req)
766        msg_short = '%s %s' % (return_string, test)
767        scene_test_summary += msg_short + '\n'
768        if test in _LIGHTING_CONTROL_TESTS and not testing_flash_with_controller:
769          print('Turn lights ON in rig and press <ENTER> to continue.')
770
771      # unit is millisecond for execution time record in CtsVerifier
772      scene_end_time = int(round(time.time() * 1000))
773      skip_string = ''
774      tot_tests = len(scene_test_list)
775      if num_skip > 0:
776        skip_string = f",{num_skip} test{'s' if num_skip > 1 else ''} skipped"
777      test_result = '%d / %d tests passed (%.1f%%)%s' % (
778          num_pass + num_not_mandated_fail, len(scene_test_list) - num_skip,
779          100.0 * float(num_pass + num_not_mandated_fail) /
780          (len(scene_test_list) - num_skip)
781          if len(scene_test_list) != num_skip else 100.0, skip_string)
782      logging.info(test_result)
783      if num_not_mandated_fail > 0:
784        logging.info('(*) %s not_yet_mandated tests failed',
785                     num_not_mandated_fail)
786
787      tot_pass += num_pass
788      logging.info('scene tests: %s, Total tests passed: %s', tot_tests,
789                   tot_pass)
790      if tot_tests > 0:
791        logging.info('%s compatibility score: %.f/100\n',
792                     s, 100 * num_pass / tot_tests)
793        scene_test_summary_path = os.path.join(mobly_scene_output_logs_path,
794                                               'scene_test_summary.txt')
795        with open(scene_test_summary_path, 'w') as f:
796          f.write(scene_test_summary)
797        results[s][RESULT_KEY] = (RESULT_PASS if num_fail == 0 else RESULT_FAIL)
798        results[s][SUMMARY_KEY] = scene_test_summary_path
799        results[s][TIME_KEY_START] = scene_start_time
800        results[s][TIME_KEY_END] = scene_end_time
801      else:
802        logging.info('%s compatibility score: 0/100\n')
803
804      # Delete temporary yml file after scene run.
805      new_yaml_file_path = os.path.join(YAML_FILE_DIR, new_yml_file_name)
806      os.remove(new_yaml_file_path)
807
808    # Log results per camera
809    logging.info('Reporting camera %s ITS results to CtsVerifier', camera_id)
810    report_result(device_id, camera_id, results)
811
812  logging.info('Test execution completed.')
813
814  # Power down tablet
815  if tablet_id:
816    cmd = f'adb -s {tablet_id} shell input keyevent KEYCODE_POWER'
817    subprocess.Popen(cmd.split())
818
819if __name__ == '__main__':
820  main()
821