blob: 883d3bd7aee15f1be16980f7061119239486a0a6 [file] [log] [blame]
# Copyright 2019 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.import json
import logging
import os
import requests
from autotest_lib.client.bin import test
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.cros import chrome
from autotest_lib.client.cros import constants as cros_constants
from autotest_lib.client.cros.multimedia import local_facade_factory
from import helper_logger
from collections import namedtuple
from string import Template
class video_AVAnalysis(test.test):
"""This test plays a video on DUT so it can be recorded.
The recording will be carried out by a recording server connected
to this DUT via HDMI cable (hence dependency on cros_av_analysis label.
The recording will then be uploaded to Google Cloud storage and analyzed
for performance and quality metrics.
version = 1
dut_info = namedtuple('dut_info', 'ip, board, build')
srv_info = namedtuple('srv_info', 'ip, port, path')
rec_info = namedtuple('rec_info', ('vid_name, duration, project_id, '
' project_name, bucket_name'))
tst_info = namedtuple('tst_info', ('check_smoothness, check_psnr, '
'check_sync, check_audio, check_color, '
'vid_id, vid_name'))
def gather_config_info(self):
"""Gathers info relevant for AVAS config file creation.
Method might seem weird, but it isolates values that at some
point could be gathered from a config or setting file instead
to provide more flexibility.
board = utils.get_platform()
if board is None:
board = utils.get_board()
self.dut_info.board = board = utils.get_chromeos_version().replace('.', '_')
self.rec_info.vid_name = '{}_{}_{}.mp4'.format(, self.dut_info.board, 'vp9')
self.rec_info.duration = '5'
self.rec_info.project_id = '40'
self.rec_info.project_name = 'cros-av'
self.rec_info.bucket_name = 'cros-av-analysis'
self.tst_info.check_smoothness = 'true'
self.tst_info.check_psnr = 'true'
self.tst_info.check_sync = 'false'
self.tst_info.check_audio = 'false'
self.tst_info.check_color = 'false'
self.tst_info.vid_id = '417'
self.tst_info.vid_name = 'cros_vp9_720_60'
def gather_runtime_info(self):
"""Gathers pieces of info required for test execution"""
self.dut_info.ip = utils.get_ip_address()
self.srv_info.ip = self.get_server_ip(self.dut_info.ip)
self.srv_info.port = '5000'
self.srv_info.path = 'record_and_upload'
def get_server_ip(self, dut_ip):
"""Returns recorder server IP address.
This method uses DUT IP to calculate IP of recording server. Note that
we rely on a protocol here, when the lab is setup, DUTs and recorders
are assigned IPs in pairs and the server is always one lower. As in,
in a V4 address, the last integer segment is less than DUTs last int
segment by 1.
@param dut_ip: IP address of DUT.
segments = dut_ip.split('.')
if len(segments) != 4:
raise Exception('IP address of DUT did not have 4 segments')
last_segment = int(segments[3])
if last_segment > 255 or last_segment < 1:
raise Exception('Value of last IP segment of DUT is invalid')
last_segment = last_segment - 1
segments[3] = str(last_segment)
server_ip = '.'.join(segments)
return server_ip
def start_recording(self):
"""Starts recording on recording server.
Makes an http POST request to the recording server to start
recording and processes the response. The body of the post
contains config file data.
destination = 'http://{}:{}/{}'.format(self.srv_info.ip,
query_params = {'filename': self.rec_info.vid_name,
'duration': self.rec_info.duration}
config_text = self.get_config_string()
headers = {'content-type': 'text/plain'}
response =, params=query_params,
data=config_text, timeout=60, headers=headers)
logging.debug('Response received is: ({}, {})'.format(
response.status_code, response.content))
if response.status_code != 200:
raise error.TestFail(
'Recording server failed with response: ({}, {})'.format(
response.status_code, response.content))
def get_config_string(self):
"""Write up config text so that AVAS can correctly process video."""
path_prefix = '/bigstore/cros-av-analysis/{}'
filepath = path_prefix.format(self.rec_info.vid_name)
config_dict = {'filepath': filepath, 'triggered_by': 'vsuley'}
config_path = os.path.join(self.bindir, 'config_template.txt')
with open(config_path) as file:
config_template = Template(
config = config_template.substitute(config_dict)
return config
def run_once(self, video, arc_mode=False):
"""Plays video on DUT for recording & analysis."""
with chrome.Chrome(
init_network_controller=True) as cr:
factory = local_facade_factory.LocalFacadeFactory(cr)
display_facade = factory.create_display_facade()
logging.debug('Setting mirrorred to True')
tab1 = cr.browser.tabs.New()