# Copyright 2024 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. """Verify feature combinations for stabilization, 10-bit, and frame rate.""" import concurrent.futures from datetime import datetime # pylint: disable=g-importing-member from google.protobuf import text_format import gc import logging import os import threading import time from mobly import test_runner import its_base_test import error_util import camera_properties_utils import capture_request_utils import gen2_rig_controller_utils import its_session_utils import lighting_control_utils import preview_processing_utils import video_processing_utils import feature_combination_info_pb2 _BIT_HLG10 = 0x01 # bit 1 for feature mask _BIT_STABILIZATION = 0x02 # bit 2 for feature mask _FPS_30_60 = (30, 60) _FPS_SELECTION_ATOL = 0.01 _FPS_ATOL_CODEC = 1.2 _FPS_ATOL_METADATA = 1.1 _VALID_RIGS = ['arduino', 'gen2_rotator'] _NAME = os.path.splitext(os.path.basename(__file__))[0] _SEC_TO_NSEC = 1_000_000_000 class FeatureCombinationTest(its_base_test.ItsBaseTest): """Tests camera feature combinations. The combination of camera features tested by this function are: - Preview stabilization - Target FPS range - HLG 10-bit HDR Camera is moved in sensor fusion rig on an arc of 15 degrees. Speed is set to mimic hand movement (and not be too fast). Preview is captured after rotation rig starts moving and the gyroscope data is dumped. Preview stabilization: The recorded preview is processed to dump all of the frames to PNG files. Camera movement is extracted from frames by determining max angle of deflection in video movement vs max angle of deflection in gyroscope movement. Test is a PASS if rotation is reduced in video. Target FPS range: The recorded preview has the expected fps range. For example, if [60, 60] is set as targetFpsRange, the camera device is expected to produce 60fps preview/video. HLG 10-bit HDR: The recorded preview has valid 10-bit HLG outputs. """ features_passed_lock = threading.Lock() def test_feature_combination(self): # Use a pool of threads to execute calls asynchronously with concurrent.futures.ThreadPoolExecutor() as executor: self._test_feature_combination(executor) def _create_feature_combo_proto(self): """Start logging feature combination info for camera in proto.""" feature_combo_for_camera = ( feature_combination_info_pb2.FeatureCombinationForCamera()) feature_combo_for_camera.camera_id = self.camera_id return feature_combo_for_camera def _stabilization_mode_to_proto_enum(self, stabilization): """Convert stabilization mode to the protobuf enum.""" if stabilization == camera_properties_utils.STABILIZATION_MODE_ON: return feature_combination_info_pb2.STABILIZATION_ON elif stabilization == camera_properties_utils.STABILIZATION_MODE_PREVIEW: return feature_combination_info_pb2.STABILIZATION_PREVIEW else: return feature_combination_info_pb2.STABILIZATION_OFF def _add_feature_combo_entry_to_proto(self, feature_combo_for_camera, output_surfaces, support_claimed, is_supported, fps_range, stabilization, failure_msg): """Log whether a feature combination is supported.""" entry = feature_combination_info_pb2.FeatureCombinationEntry() entry.is_supported = is_supported entry.support_claimed = support_claimed entry.failure_message = failure_msg for surface in output_surfaces: config_entry = feature_combination_info_pb2.OutputConfiguration() config_entry.image_format = surface['format_code'] config_entry.size.width = surface['width'] config_entry.size.height = surface['height'] config_entry.stream_usecase = feature_combination_info_pb2.USECASE_DEFAULT config_entry.dynamic_range_profile = ( feature_combination_info_pb2.PROFILE_HLG10 if surface['hlg10'] else feature_combination_info_pb2.PROFILE_STANDARD) entry.session_configuration.output_configurations.append(config_entry) entry.session_configuration.stabilization = ( self._stabilization_mode_to_proto_enum(stabilization) ) entry.session_configuration.frame_rate_range.max = fps_range[1] entry.session_configuration.frame_rate_range.min = fps_range[0] feature_combo_for_camera.entries.append(entry) def _output_feature_combo_proto(self, feature_combo_for_camera): """Finish logging feature combination info and write to ReportLogFiles.""" database = feature_combination_info_pb2.FeatureCombinationDatabase() database.build_fingerprint = ( its_session_utils.get_build_fingerprint(self.dut.serial)) database.timestamp_in_sec = int(time.time()) database.feature_combination_for_camera.append(feature_combo_for_camera) current_time = datetime.now().strftime('%Y_%m_%d_%H_%M_%S') proto_file_name = ( f'{self.dut.serial}_camera_{self.camera_id}_{current_time}.pb' ) logging.debug('proto_file_name %s', proto_file_name) logging.debug('root_output_path %s', self.root_output_path) proto_file_path = os.path.join(self.root_output_path, proto_file_name) with open(proto_file_path, 'wb') as f: f.write(database.SerializeToString()) txtpb_file_name = proto_file_name.replace('.pb', '.txtpb') txtpb_file_path = os.path.join(self.root_output_path, txtpb_file_name) with open(txtpb_file_path, 'w') as tf: database_str = text_format.MessageToString(database, as_one_line=True) tf.write(database_str) print(f'feature_query_proto:{txtpb_file_name}') def _finish_combination( self, combination_name, stabilize_mode, support_claimed, passed, recording_obj, gyro_events, test_name, log_path, facing, output_surfaces, fps_range, hlg10, features_passed, streams_name, fps_range_tuple, failure_msg ): """Finish verifying a feature combo & preview stabilization if necessary.""" result = {'name': combination_name, 'output_surfaces': output_surfaces, 'fps_range': fps_range, 'stabilize_mode': stabilize_mode, 'support_claimed': support_claimed, 'passed': passed, 'failure_msg': failure_msg} is_stabilized = ( stabilize_mode != camera_properties_utils.STABILIZATION_MODE_OFF ) if is_stabilized: stabilization_result = ( preview_processing_utils.verify_stabilization( recording_obj, gyro_events, test_name, log_path, facing ) ) if stabilization_result['failure']: result['stabilization_failure'] = stabilization_result['failure'] result['passed'] = False if result['passed']: with self.features_passed_lock: its_session_utils.mark_features_passed( features_passed, streams_name, fps_range_tuple, hlg10, is_stabilized) # Remove video clip if test passes to save space try: file_name = recording_obj['recordedOutputPath'].split('/')[-1] os.remove(os.path.join(log_path, file_name)) except FileNotFoundError: logging.debug('File Not Found: %s', str(file_name)) return result def _append_test_failure(self, failures, support_claimed, combination, msg): msg_for_combination = f'{combination}:{msg}' if (support_claimed == feature_combination_info_pb2.SUPPORT_YES): failures['required'].append(msg_for_combination) else: failures['optional'].append(msg_for_combination) def _handle_one_verification_result( self, result, test_failures, database): """Handle the verification result for one combination. This function appends test failure for video stabilization and store the verification result to protobuf. """ logging.debug('Verification result: %s', result) if 'stabilization_failure' in result: result['failure_msg'] += result['stabilization_failure'] self._append_test_failure( test_failures, result['support_claimed'], result['name'], result['stabilization_failure'] ) self._add_feature_combo_entry_to_proto( database, result['output_surfaces'], result['support_claimed'], result['passed'], result['fps_range'], result['stabilize_mode'], result['failure_msg'] ) def _handle_one_completed_future( self, pending_futures, future, test_failures, database): result = future.result() pending_futures.remove(future) self._handle_one_verification_result(result, test_failures, database) gc.collect() def _handle_completed_futures(self, verifications, test_failures, database): for future in verifications[:]: if future.done(): self._handle_one_completed_future( verifications, future, test_failures, database) def _drain_feature_verification_futures( self, verifications, test_failures, database): for future in concurrent.futures.as_completed(verifications): self._handle_one_completed_future( verifications, future, test_failures, database) def _setup_lighting_cntl(self): if self.lighting_cntl == 'gen2_lights': lights_port = gen2_rig_controller_utils.find_serial_port( self.lighting_cntl) lights_channel = int(self.lighting_ch) gen2_rig_controller_utils.set_lighting_state( lights_port, lights_channel, 'ON') elif self.lighting_cntl == 'arduino': lights_channel = int(self.lighting_ch) lights_port = lighting_control_utils.lighting_control( self.lighting_cntl, lights_channel) lighting_control_utils.set_lighting_state( lights_port, lights_channel, 'ON') def _test_feature_combination(self, executor): """Tests features using an injected ThreadPoolExecutor for analysis. Args: executor: a ThreadPoolExecutor to analyze recordings asynchronously. """ rot_rig = {} log_path = self.log_path with its_session_utils.ItsSession( device_id=self.dut.serial, camera_id=self.camera_id) as cam: # Skip if not primary camera camera_properties_utils.skip_unless( cam.is_primary_camera()) # Skip if the device doesn't support feature combination query props = cam.get_camera_properties() feature_combination_query_version = props.get( 'android.info.sessionConfigurationQueryVersion') if not feature_combination_query_version: feature_combination_query_version = ( its_session_utils.ANDROID14_API_LEVEL ) support_query = (feature_combination_query_version >= its_session_utils.ANDROID15_API_LEVEL) # Log ffmpeg version being used video_processing_utils.log_ffmpeg_version() # Raise error if not FRONT or REAR facing camera facing = props['android.lens.facing'] camera_properties_utils.check_front_or_rear_camera(props) # Stabilization: Make sure to test ON first in order to be able to skip # OFF stabilization_params = [True, False] stabilization_modes = props[ 'android.control.availableVideoStabilizationModes'] # Initialize rotation rig rot_rig['cntl'] = self.rotator_cntl rot_rig['ch'] = self.rotator_ch if (rot_rig['cntl'].lower() not in _VALID_RIGS and len(stabilization_modes) > 1): raise AssertionError( f'You must use the arduino or gen2_rotator controller for {_NAME}.') # Initialize lighting rig self._setup_lighting_cntl() # List of queryable stream combinations combinations_str, combinations = cam.get_queryable_stream_combinations() logging.debug('Queryable stream combinations: %s', combinations_str) configs = props['android.scaler.streamConfigurationMap'][ 'availableStreamConfigurations'] fps_ranges = camera_properties_utils.get_ae_target_fps_ranges(props) fps_params = [fps for fps in fps_ranges if ( fps[1] in _FPS_30_60)] hlg10_params = [True, False] # 'required': failed combinations that are required to pass # 'optional': failed combinations that are used for analytics only test_failures = {'required': [], 'optional': []} feature_verification_futures = [] database = self._create_feature_combo_proto() features_passed = {} for fps_range in fps_params: fps_range_tuple = tuple(fps_range) for hlg10 in hlg10_params: for is_stabilized in stabilization_params: for stream_combination in combinations: streams_name = stream_combination['name'] combo_version = stream_combination['version'] combination_name = (f'(streams: {streams_name}, hlg10: {hlg10}, ' f'stabilization: {is_stabilized}, fps_range: ' f'[{fps_range[0]}, {fps_range[1]}])') min_frame_duration = 0 configured_streams = [] if (stream_combination['combination'][0]['format'] != its_session_utils.PRIVATE_FORMAT): raise AssertionError( f'First stream for {streams_name} must be PRIV') video_stream_index = 0 for index, stream in enumerate(stream_combination['combination']): fmt = None size = [int(e) for e in stream['size'].split('x')] match stream['format']: case its_session_utils.PRIVATE_FORMAT: fmt = capture_request_utils.FMT_CODE_PRIV video_stream_index = index case 'jpeg': fmt = capture_request_utils.FMT_CODE_JPEG case its_session_utils.JPEG_R_FMT_STR: fmt = capture_request_utils.FMT_CODE_JPEG_R case 'yuv': fmt = capture_request_utils.FMT_CODE_YUV case _: raise AssertionError( f'Unsupported stream format {stream["format"]}') config = [x for x in configs if x['format'] == fmt and x['width'] == size[0] and x['height'] == size[1]] min_frame_duration = max( config[0]['minFrameDuration'], min_frame_duration) logging.debug( 'format is %s, min_frame_duration is %d}', stream['format'], config[0]['minFrameDuration']) configured_streams.append( {'formatStr': stream['format'], 'format': fmt, 'width': size[0], 'height': size[1]}) # Check if the FPS range is supported max_achievable_fps = _SEC_TO_NSEC / min_frame_duration if max_achievable_fps < fps_range[1] - _FPS_SELECTION_ATOL: continue # Check if the hlg10 is supported for the video size and fps video_size = ( stream_combination['combination'][video_stream_index]['size'] ) if (hlg10 and not cam.is_hlg10_recording_supported_for_size_and_fps( video_size, fps_range[1]) ): continue # Check if stabilization is supported: Use video stabilization # if there is a dedicated video stream if not is_stabilized: stabilize_mode = camera_properties_utils.STABILIZATION_MODE_OFF elif video_stream_index == 0: stabilize_mode = ( camera_properties_utils.STABILIZATION_MODE_PREVIEW ) else: stabilize_mode = camera_properties_utils.STABILIZATION_MODE_ON if stabilize_mode not in stabilization_modes: continue logging.debug('combination name: %s', combination_name) # Construct output surfaces output_surfaces = [] for index, configured_stream in enumerate(configured_streams): hlg10_stream = (configured_stream['formatStr'] == its_session_utils.PRIVATE_FORMAT and hlg10) is_video = index == video_stream_index output_surfaces.append( {'format': configured_stream['formatStr'], 'format_code': configured_stream['format'], 'width': configured_stream['width'], 'height': configured_stream['height'], 'hlg10': hlg10_stream, 'is_video': is_video} ) settings = { 'android.control.videoStabilizationMode': stabilize_mode, 'android.control.aeTargetFpsRange': fps_range, } # UNKNOWN means that the device doesn't support # is_stream_combination_supported (with session parameters) for # this combination of features. support_claimed = feature_combination_info_pb2.SUPPORT_UNKNOWN if feature_combination_query_version >= combo_version: # Is the feature combination supported? if cam.is_stream_combination_supported( output_surfaces, settings): support_claimed = feature_combination_info_pb2.SUPPORT_YES else: support_claimed = feature_combination_info_pb2.SUPPORT_NO logging.debug('%s not supported', combination_name) passed = True combined_failure_msg = "" # If a superset of features are already tested, skip and assuming # the subset of those features are supported. Do not skip [60, *] # even if its superset feature passes (b/385753212). with self.features_passed_lock: skip_test = ( streams_name in features_passed and fps_range_tuple in features_passed[streams_name] and fps_range[0] < 60 and its_session_utils.check_features_passed( features_passed, streams_name, fps_range_tuple, hlg10, is_stabilized) ) if skip_test: self._add_feature_combo_entry_to_proto( database, output_surfaces, support_claimed, passed, fps_range, stabilize_mode, combined_failure_msg) continue # In case collect_data_with_surfaces throws an exception, treat it # as an unsupported combination. (b/385753212#2) try: recording_obj = ( preview_processing_utils.collect_data_with_surfaces( cam, self.tablet_device, output_surfaces, video_stream_index, stabilize_mode, rot_rig=rot_rig, fps_range=fps_range ) ) except error_util.CameraItsError as e: if (support_query and support_claimed == feature_combination_info_pb2.SUPPORT_YES ): raise e failure_msg = f'collect_data_with_surfaces throws exception: {e}' logging.debug(failure_msg) self._append_test_failure( test_failures, support_claimed, combination_name, failure_msg ) passed = False combined_failure_msg += f'{failure_msg};' self._add_feature_combo_entry_to_proto( database, output_surfaces, support_claimed, passed, fps_range, stabilize_mode, combined_failure_msg) cam.reset_socket_and_camera() continue gyro_events = None if is_stabilized: # Get gyro events logging.debug('Reading out inertial sensor events') gyro_events = cam.get_sensor_events()['gyro'] logging.debug('Number of gyro samples %d', len(gyro_events)) # Grab the video from the file location on DUT self.dut.adb.pull([recording_obj['recordedOutputPath'], log_path]) # Delete the video file from the DUT self.dut.adb.shell('rm %s' % recording_obj['recordedOutputPath']) # Verify FPS by inspecting the video clip preview_file_name = ( recording_obj['recordedOutputPath'].split('/')[-1]) preview_file_name_with_path = os.path.join( self.log_path, preview_file_name) avg_frame_rate_codec = ( video_processing_utils.get_avg_frame_rate( preview_file_name_with_path)) logging.debug('Average codec frame rate for %s is %f', combination_name, avg_frame_rate_codec) if (avg_frame_rate_codec > fps_range[1] + _FPS_ATOL_CODEC or avg_frame_rate_codec < fps_range[0] - _FPS_ATOL_CODEC): failure_msg = ( f'Average video clip frame rate {avg_frame_rate_codec} ' f'exceeding the allowed range of ' f'({fps_range[0]}-{_FPS_ATOL_CODEC}, ' f'{fps_range[1]}+{_FPS_ATOL_CODEC})') combined_failure_msg += f'{failure_msg};' self._append_test_failure( test_failures, support_claimed, combination_name, failure_msg ) passed = False # Verify FPS by inspecting the result metadata capture_results = recording_obj['captureMetadata'] if len(capture_results) <= 1: raise AssertionError( f'{combination_name}: captureMetadata has only ' f'{len(capture_results)} frames') last_t = capture_results[-1]['android.sensor.timestamp'] first_t = capture_results[0]['android.sensor.timestamp'] avg_frame_duration = ( (last_t - first_t) / (len(capture_results) - 1)) avg_frame_rate_metadata = _SEC_TO_NSEC / avg_frame_duration logging.debug('Average metadata frame rate for %s is %f', combination_name, avg_frame_rate_metadata) if (avg_frame_rate_metadata > fps_range[1] + _FPS_ATOL_METADATA or avg_frame_rate_metadata < fps_range[0] - _FPS_ATOL_METADATA): failure_msg = ( f'Average frame rate {avg_frame_rate_metadata} ' f'exceeding the allowed range of ' f'({fps_range[0]}-{_FPS_ATOL_METADATA}, ' f'{fps_range[1]}+{_FPS_ATOL_METADATA})') combined_failure_msg += f'{failure_msg};' self._append_test_failure( test_failures, support_claimed, combination_name, failure_msg ) passed = False # Verify color space color_space = video_processing_utils.get_video_colorspace( self.log_path, preview_file_name_with_path) if (hlg10 and video_processing_utils.COLORSPACE_HDR not in color_space): failure_msg = ( f'video color space {color_space} ' 'is missing COLORSPACE_HDR') combined_failure_msg += f'{failure_msg};' self._append_test_failure( test_failures, support_claimed, combination_name, failure_msg ) passed = False if not self.parallel_execution: verification_result = self._finish_combination( combination_name, stabilize_mode, support_claimed, passed, recording_obj, gyro_events, _NAME, log_path, facing, output_surfaces, fps_range, hlg10, features_passed, streams_name, fps_range_tuple, combined_failure_msg ) self._handle_one_verification_result( verification_result, test_failures, database) else: future = executor.submit( self._finish_combination, combination_name, stabilize_mode, support_claimed, passed, recording_obj, gyro_events, _NAME, log_path, facing, output_surfaces, fps_range, hlg10, features_passed, streams_name, fps_range_tuple, combined_failure_msg ) # Handle completed feature verification futures self._handle_completed_futures( feature_verification_futures, test_failures, database) feature_verification_futures.append(future) # Drain the remaining feature combination results self._drain_feature_verification_futures( feature_verification_futures, test_failures, database) # Output the feature combination proto to ItsService and optionally to # file self._output_feature_combo_proto(database) # Assert PASS/FAIL criteria if test_failures: logging.debug(test_failures) if test_failures['required']: raise AssertionError(test_failures['required']) if __name__ == '__main__': test_runner.main()