1# Copyright 2019 The Chromium OS Authors. All rights reserved. 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file.import json 4 5import logging 6import os 7import requests 8 9from autotest_lib.client.bin import test 10from autotest_lib.client.bin import utils 11from autotest_lib.client.common_lib import error 12from autotest_lib.client.common_lib.cros import chrome 13from autotest_lib.client.cros import constants as cros_constants 14from autotest_lib.client.cros.multimedia import local_facade_factory 15from autotest_lib.client.cros.video import helper_logger 16from collections import namedtuple 17from string import Template 18 19 20class video_AVAnalysis(test.test): 21 """This test plays a video on DUT so it can be recorded. 22 23 The recording will be carried out by a recording server connected 24 to this DUT via HDMI cable (hence dependency on cros_av_analysis label. 25 The recording will then be uploaded to Google Cloud storage and analyzed 26 for performance and quality metrics. 27 """ 28 version = 1 29 dut_info = namedtuple('dut_info', 'ip, board, build') 30 srv_info = namedtuple('srv_info', 'ip, port, path') 31 rec_info = namedtuple('rec_info', ('vid_name, duration, project_id, ' 32 ' project_name, bucket_name')) 33 tst_info = namedtuple('tst_info', ('check_smoothness, check_psnr, ' 34 'check_sync, check_audio, check_color, ' 35 'vid_id, vid_name')) 36 37 def gather_config_info(self): 38 """Gathers info relevant for AVAS config file creation. 39 40 Method might seem weird, but it isolates values that at some 41 point could be gathered from a config or setting file instead 42 to provide more flexibility. 43 """ 44 board = utils.get_platform() 45 if board is None: 46 board = utils.get_board() 47 self.dut_info.board = board 48 self.dut_info.build = utils.get_chromeos_version().replace('.', '_') 49 50 self.rec_info.vid_name = '{}_{}_{}.mp4'.format( 51 self.dut_info.build, self.dut_info.board, 'vp9') 52 self.rec_info.duration = '5' 53 self.rec_info.project_id = '40' 54 self.rec_info.project_name = 'cros-av' 55 self.rec_info.bucket_name = 'cros-av-analysis' 56 57 self.tst_info.check_smoothness = 'true' 58 self.tst_info.check_psnr = 'true' 59 self.tst_info.check_sync = 'false' 60 self.tst_info.check_audio = 'false' 61 self.tst_info.check_color = 'false' 62 self.tst_info.vid_id = '417' 63 self.tst_info.vid_name = 'cros_vp9_720_60' 64 65 def gather_runtime_info(self): 66 """Gathers pieces of info required for test execution""" 67 self.dut_info.ip = utils.get_ip_address() 68 self.srv_info.ip = self.get_server_ip(self.dut_info.ip) 69 logging.debug('----------I-P------------') 70 logging.debug(self.dut_info.ip) 71 logging.debug(self.srv_info.ip) 72 self.srv_info.port = '5000' 73 self.srv_info.path = 'record_and_upload' 74 75 def get_server_ip(self, dut_ip): 76 """Returns recorder server IP address. 77 78 This method uses DUT IP to calculate IP of recording server. Note that 79 we rely on a protocol here, when the lab is setup, DUTs and recorders 80 are assigned IPs in pairs and the server is always one lower. As in, 81 in a V4 address, the last integer segment is less than DUTs last int 82 segment by 1. 83 84 @param dut_ip: IP address of DUT. 85 """ 86 segments = dut_ip.split('.') 87 if len(segments) != 4: 88 raise Exception('IP address of DUT did not have 4 segments') 89 last_segment = int(segments[3]) 90 if last_segment > 255 or last_segment < 1: 91 raise Exception('Value of last IP segment of DUT is invalid') 92 93 last_segment = last_segment - 1 94 segments[3] = str(last_segment) 95 server_ip = '.'.join(segments) 96 return server_ip 97 98 def start_recording(self): 99 """Starts recording on recording server. 100 101 Makes an http POST request to the recording server to start 102 recording and processes the response. The body of the post 103 contains config file data. 104 """ 105 destination = 'http://{}:{}/{}'.format(self.srv_info.ip, 106 self.srv_info.port, 107 self.srv_info.path) 108 query_params = {'filename': self.rec_info.vid_name, 109 'duration': self.rec_info.duration} 110 config_text = self.get_config_string() 111 headers = {'content-type': 'text/plain'} 112 response = requests.post(destination, params=query_params, 113 data=config_text, timeout=60, headers=headers) 114 logging.debug('Response received is: ({}, {})'.format( 115 response.status_code, response.content)) 116 117 if response.status_code != 200: 118 raise error.TestFail( 119 'Recording server failed with response: ({}, {})'.format( 120 response.status_code, response.content)) 121 122 def get_config_string(self): 123 """Write up config text so that AVAS can correctly process video.""" 124 path_prefix = '/bigstore/cros-av-analysis/{}' 125 filepath = path_prefix.format(self.rec_info.vid_name) 126 config_dict = {'filepath': filepath, 'triggered_by': 'vsuley'} 127 config_dict.update(vars(self.rec_info)) 128 config_dict.update(vars(self.dut_info)) 129 config_dict.update(vars(self.tst_info)) 130 131 config_path = os.path.join(self.bindir, 'config_template.txt') 132 with open(config_path) as file: 133 config_template = Template(file.read()) 134 config = config_template.substitute(config_dict) 135 return config 136 137 def run_once(self, video, arc_mode=False): 138 """Plays video on DUT for recording & analysis.""" 139 self.gather_config_info() 140 self.gather_runtime_info() 141 with chrome.Chrome( 142 extra_browser_args=helper_logger.chrome_vmodule_flag(), 143 extension_paths=[cros_constants.DISPLAY_TEST_EXTENSION], 144 autotest_ext=True, 145 arc_mode="disabled", 146 init_network_controller=True) as cr: 147 factory = local_facade_factory.LocalFacadeFactory(cr) 148 display_facade = factory.create_display_facade() 149 logging.debug('Setting mirrorred to True') 150 display_facade.set_mirrored(True) 151 display_facade.set_fullscreen(True) 152 tab1 = cr.browser.tabs.New() 153 tab1.Navigate(video) 154 tab1.WaitForDocumentReadyStateToBeComplete() 155 self.start_recording() 156