1# Copyright 2016 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 logging
16import os
17import os.path
18import shutil
19import subprocess
20import sys
21import tempfile
22
23import numpy as np
24
25import run_all_tests  # from same tools directory as run_sensor_fusion.py
26
27_CORR_DIST_THRESH_MAX = 0.005  # must match value in test_sensor_fusion.py
28_NUM_RUNS = 1
29_TEST_BED_SENSOR_FUSION = 'TEST_BED_SENSOR_FUSION'
30_TIME_SHIFT_MATCH = 'Best correlation of '
31
32
33def find_time_shift(out_file_path):
34  """Search through a test run's test_log.DEBUG for the best time shift.
35
36  Args:
37    out_file_path: File path for stdout logs to search through
38
39  Returns:
40    Float num of best time shift, if one is found. Otherwise, None.
41  """
42  line = find_matching_line(out_file_path, _TIME_SHIFT_MATCH)
43  if line is None:
44    return None
45  else:
46    words = line.split(' ')
47    time_shift = float(words[-1][:-3])  # strip off 'ms'
48    fit_corr = float(words[-5])
49    return {'time_shift': time_shift, 'corr': fit_corr}
50
51
52def find_matching_line(file_path, match_string):
53  """Search each line in the file at 'file_path' for match_string.
54
55  Args:
56      file_path: File path for file being searched
57      match_string: Sting used to match against lines
58
59  Returns:
60      The first matching line. If none exists, returns None.
61  """
62  with open(file_path) as f:
63    for line in f:
64      if match_string in line:
65        return line
66  return None
67
68
69def main():
70  """Run the sensor_fusion test for stastical purposes.
71
72    Script should be run from the top-level CameraITS directory.
73    All parameters expect 'num_runs' are defined in config.yml.
74    num_runs is defined at run time with 'num_runs=<int>'
75    'camera_id' can be over-written at command line to allow different
76    camera_ids facing the same direction to be tested.
77
78    ie. python tools/run_all_tests.py num_runs=10  # n=10 w/ config.yml cam
79        python tools/run_all_tests.py camera=0 num_runs=10  # n=10 w/ cam[0]
80        python tools/run_all_tests.py camera=0.4 num_runs=10 # n=10 w/ cam[0.4]
81
82    Command line arguments:
83        camera_id: camera_id or list of camera_ids.
84        num_runs: integer number of runs to get statistical values
85
86    All other config values are stored in config.yml file.
87  """
88  logging.basicConfig(level=logging.INFO)
89  # Make output directories to hold the generated files.
90  topdir = tempfile.mkdtemp(prefix='CameraITS_')
91  subprocess.call(['chmod', 'g+rx', topdir])
92
93  camera_id_combos = []
94
95  # Override camera with cmd line values if available
96  num_runs = _NUM_RUNS
97  get_argv_vals = lambda x: x.split('=')[1]
98  for s in list(sys.argv[1:]):
99    if 'camera=' in s:
100      camera_id_combos = str(get_argv_vals(s)).split(',')
101    elif 'num_runs=' in s:
102      num_runs = int(get_argv_vals(s))
103
104  # Read config file and extract relevant TestBed
105  config_file_contents = run_all_tests.get_config_file_contents()
106  for i in config_file_contents['TestBeds']:
107    if i['Name'] != _TEST_BED_SENSOR_FUSION:
108      config_file_contents['TestBeds'].remove(i)
109
110  # Get test parameters from config file
111  test_params_content = run_all_tests.get_test_params(config_file_contents)
112  if not camera_id_combos:
113    camera_id_combos = test_params_content['camera'].split(',')
114  debug = test_params_content['debug_mode']
115  fps = test_params_content['fps']
116  img_size = test_params_content['img_size']
117
118  # Get dut id
119  device_id = run_all_tests.get_device_serial_number(
120      'dut', config_file_contents)
121
122  # Log run info
123  logging.info('Running sensor_fusion on device: %s, camera: %s',
124               device_id, camera_id_combos)
125  logging.info('Saving output files to: %s', topdir)
126
127  for camera_id in camera_id_combos:
128    time_shifts = []
129    # A subdir in topdir will be created for each camera_id.
130    test_params_content['camera'] = camera_id
131    test_params_content['scene'] = 'sensor_fusion'
132    config_file_contents['TestBeds'][0]['TestParams'] = test_params_content
133    os.mkdir(os.path.join(topdir, camera_id))
134
135    # Add the MoblyParams to config.yml file store camera_id test results.
136    mobly_output_logs_path = os.path.join(topdir, camera_id)
137    mobly_scene_output_logs_path = os.path.join(
138        mobly_output_logs_path, 'sensor_fusion')
139    mobly_params_dict = {
140        'MoblyParams': {
141            'LogPath': mobly_scene_output_logs_path
142        }
143    }
144    config_file_contents.update(mobly_params_dict)
145    logging.debug('Config file contents: %s', config_file_contents)
146    tmp_yml_file_name = run_all_tests.get_updated_yml_file(config_file_contents)
147    logging.info('Using %s as temporary config yml file', tmp_yml_file_name)
148
149    # Run tests
150    logging.info('%d runs for test_sensor_fusion with camera %s',
151                 num_runs, camera_id)
152    logging.info('FPS: %d, img size: %s', fps, img_size)
153    for _ in range(num_runs):
154      cmd = ['python',
155             os.path.join(os.environ['CAMERA_ITS_TOP'], 'tests',
156                          'sensor_fusion', 'test_sensor_fusion.py'),
157             '-c',
158             f'{tmp_yml_file_name}'
159             ]
160      # pylint: disable=subprocess-run-check
161      with open(run_all_tests.MOBLY_TEST_SUMMARY_TXT_FILE, 'w') as fp:
162        output = subprocess.run(cmd, stdout=fp)
163      # pylint: enable=subprocess-run-check
164
165      with open(run_all_tests.MOBLY_TEST_SUMMARY_TXT_FILE, 'r') as _:
166        if output.returncode == 0:
167          return_string = 'PASS'
168        else:
169          return_string = 'FAIL'
170
171        os.remove(run_all_tests.MOBLY_TEST_SUMMARY_TXT_FILE)
172        file_name = os.path.join(
173            mobly_scene_output_logs_path, _TEST_BED_SENSOR_FUSION, 'latest',
174            'test_log.DEBUG')
175        time_shift = find_time_shift(file_name)
176        logging.info('%s time_shift: %.4f ms, corr: %.6f', return_string,
177                     time_shift['time_shift'], time_shift['corr'])
178        if time_shift['corr'] < _CORR_DIST_THRESH_MAX:
179          time_shifts.append(time_shift)
180        else:
181          logging.info('Correlation distance too large. Not used for stats.')
182
183    # Summarize results with stats
184    times = [t['time_shift'] for t in time_shifts]
185    logging.info('runs: %d, time_shift mean: %.4f, sigma: %.4f',
186                 len(times), np.mean(times), np.std(times))
187
188    # Delete temporary yml file after run.
189    tmp_yml_file = os.path.join(run_all_tests.YAML_FILE_DIR, tmp_yml_file_name)
190    os.remove(tmp_yml_file)
191
192  # Delete temporary image files after run.
193  if debug == 'False':
194    logging.info('Removing tmp dir %s to save space.', topdir)
195    shutil.rmtree(topdir)
196
197  logging.info('Test completed.')
198if __name__ == '__main__':
199  main()
200
201