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