blob: c6b51f23accb632d185355fcb19d199fa64f98a5 [file] [log] [blame]
# Lint as: python2, python3
# Copyright 2014 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.
"""Facade to access the display-related functionality."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import logging
import multiprocessing
import numpy
import os
import re
import shutil
import time
import json
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib import utils as common_utils
from autotest_lib.client.common_lib.cros import retry
from autotest_lib.client.cros import constants
from autotest_lib.client.cros.graphics import graphics_utils
from autotest_lib.client.cros.multimedia import facade_resource
from autotest_lib.client.cros.multimedia import image_generator
from autotest_lib.client.cros.power import sys_power
from six.moves import range
from telemetry.internal.browser import web_contents
class TimeoutException(Exception):
"""Timeout Exception class."""
pass
_FLAKY_CALL_RETRY_TIMEOUT_SEC = 60
_FLAKY_DISPLAY_CALL_RETRY_DELAY_SEC = 2
_retry_display_call = retry.retry(
(KeyError, error.CmdError),
timeout_min=_FLAKY_CALL_RETRY_TIMEOUT_SEC / 60.0,
delay_sec=_FLAKY_DISPLAY_CALL_RETRY_DELAY_SEC)
class DisplayFacadeLocal(object):
"""Facade to access the display-related functionality.
The methods inside this class only accept Python core types.
"""
CALIBRATION_IMAGE_PATH = '/tmp/calibration.png'
MINIMUM_REFRESH_RATE_EXPECTED = 25.0
DELAY_TIME = 3
MAX_TYPEC_PORT = 6
def __init__(self, resource):
"""Initializes a DisplayFacadeLocal.
@param resource: A FacadeResource object.
"""
self._resource = resource
self._image_generator = image_generator.ImageGenerator()
@facade_resource.retry_chrome_call
def get_display_info(self):
"""Gets the display info from Chrome.system.display API.
@return array of dict for display info.
"""
extension = self._resource.get_extension(
constants.DISPLAY_TEST_EXTENSION)
extension.ExecuteJavaScript('window.__display_info = null;')
extension.ExecuteJavaScript(
"chrome.system.display.getInfo(function(info) {"
"window.__display_info = info;})")
utils.wait_for_value(lambda: (
extension.EvaluateJavaScript("window.__display_info") != None),
expected_value=True)
return extension.EvaluateJavaScript("window.__display_info")
@facade_resource.retry_chrome_call
def get_window_info(self):
"""Gets the current window info from Chrome.system.window API.
@return a dict for the information of the current window.
"""
extension = self._resource.get_extension()
extension.ExecuteJavaScript('window.__window_info = null;')
extension.ExecuteJavaScript(
"chrome.windows.getCurrent(function(info) {"
"window.__window_info = info;})")
utils.wait_for_value(lambda: (
extension.EvaluateJavaScript("window.__window_info") != None),
expected_value=True)
return extension.EvaluateJavaScript("window.__window_info")
@facade_resource.retry_chrome_call
def create_window(self, url='chrome://newtab'):
"""Creates a new window from chrome.windows.create API.
@param url: Optional URL for the new window.
@return Identifier for the new window.
@raise TimeoutException if it fails.
"""
extension = self._resource.get_extension()
extension.ExecuteJavaScript(
"""
var __new_window_id = null;
chrome.windows.create(
{url: '%s'},
function(win) {
__new_window_id = win.id});
""" % (url)
)
extension.WaitForJavaScriptCondition(
"__new_window_id !== null",
timeout=web_contents.DEFAULT_WEB_CONTENTS_TIMEOUT)
return extension.EvaluateJavaScript("__new_window_id")
@facade_resource.retry_chrome_call
def update_window(self, window_id, state=None, bounds=None):
"""Updates an existing window using the chrome.windows.update API.
@param window_id: Identifier for the window to update.
@param state: Optional string to set the state such as 'normal',
'maximized', or 'fullscreen'.
@param bounds: Optional dictionary with keys top, left, width, and
height to reposition the window.
@return True if success.
@raise TimeoutException if it fails.
"""
extension = self._resource.get_extension()
params = {}
if state:
params['state'] = state
if bounds:
params['top'] = bounds['top']
params['left'] = bounds['left']
params['width'] = bounds['width']
params['height'] = bounds['height']
if not params:
logging.info('Nothing to update for window_id={}'.format(window_id))
return True
extension.ExecuteJavaScript(
"""
var __status = 'Running';
chrome.windows.update(%d, %s,
function(win) {
__status = 'Done'});
""" % (window_id, json.dumps(params))
)
extension.WaitForJavaScriptCondition(
"__status == 'Done'",
timeout=web_contents.DEFAULT_WEB_CONTENTS_TIMEOUT)
return True
def _get_display_by_id(self, display_id):
"""Gets a display by ID.
@param display_id: id of the display.
@return: A dict of various display info.
"""
for display in self.get_display_info():
if display['id'] == display_id:
return display
raise RuntimeError('Cannot find display ' + display_id)
def get_display_modes(self, display_id):
"""Gets all the display modes for the specified display.
@param display_id: id of the display to get modes from.
@return: A list of DisplayMode dicts.
"""
display = self._get_display_by_id(display_id)
return display['modes']
def get_display_rotation(self, display_id):
"""Gets the display rotation for the specified display.
@param display_id: id of the display to get modes from.
@return: Degree of rotation.
"""
display = self._get_display_by_id(display_id)
return display['rotation']
def get_display_notifications(self):
"""Gets the display notifications
@return: Returns a list of display related notifications only.
"""
display_notifications = []
for notification in self._resource.get_visible_notifications():
if notification['id'] == 'chrome://settings/display':
display_notifications.append(notification)
return display_notifications
def set_display_rotation(self, display_id, rotation,
delay_before_rotation=0, delay_after_rotation=0):
"""Sets the display rotation for the specified display.
@param display_id: id of the display to get modes from.
@param rotation: degree of rotation
@param delay_before_rotation: time in second for delay before rotation
@param delay_after_rotation: time in second for delay after rotation
"""
time.sleep(delay_before_rotation)
extension = self._resource.get_extension(
constants.DISPLAY_TEST_EXTENSION)
extension.ExecuteJavaScript(
"""
window.__set_display_rotation_has_error = null;
chrome.system.display.setDisplayProperties('%(id)s',
{"rotation": %(rotation)d}, () => {
if (chrome.runtime.lastError) {
console.error('Failed to set display rotation',
chrome.runtime.lastError);
window.__set_display_rotation_has_error = "failure";
} else {
window.__set_display_rotation_has_error = "success";
}
});
"""
% {'id': display_id, 'rotation': rotation}
)
utils.wait_for_value(lambda: (
extension.EvaluateJavaScript(
'window.__set_display_rotation_has_error') != None),
expected_value=True)
time.sleep(delay_after_rotation)
result = extension.EvaluateJavaScript(
'window.__set_display_rotation_has_error')
if result != 'success':
raise RuntimeError('Failed to set display rotation: %r' % result)
def get_available_resolutions(self, display_id):
"""Gets the resolutions from the specified display.
@return a list of (width, height) tuples.
"""
display = self._get_display_by_id(display_id)
modes = display['modes']
if 'widthInNativePixels' not in modes[0]:
raise RuntimeError('Cannot find widthInNativePixels attribute')
if display['isInternal']:
logging.info("Getting resolutions of internal display")
return list(set([(mode['width'], mode['height']) for mode in
modes]))
return list(set([(mode['widthInNativePixels'],
mode['heightInNativePixels']) for mode in modes]))
def has_internal_display(self):
"""Returns whether the device has an internal display.
@return whether the device has an internal display
"""
return len([d for d in self.get_display_info() if d['isInternal']]) > 0
def get_internal_display_id(self):
"""Gets the internal display id.
@return the id of the internal display.
"""
for display in self.get_display_info():
if display['isInternal']:
return display['id']
raise RuntimeError('Cannot find internal display')
def get_first_external_display_id(self):
"""Gets the first external display id.
@return the id of the first external display; -1 if not found.
"""
# Get the first external and enabled display
for display in self.get_display_info():
if display['isEnabled'] and not display['isInternal']:
return display['id']
return -1
def set_resolution(self, display_id, width, height, timeout=3):
"""Sets the resolution of the specified display.
@param display_id: id of the display to set resolution for.
@param width: width of the resolution
@param height: height of the resolution
@param timeout: maximal time in seconds waiting for the new resolution
to settle in.
@raise TimeoutException when the operation is timed out.
"""
extension = self._resource.get_extension(
constants.DISPLAY_TEST_EXTENSION)
extension.ExecuteJavaScript(
"""
window.__set_resolution_progress = null;
chrome.system.display.getInfo((info_array) => {
var mode;
for (var info of info_array) {
if (info['id'] == '%(id)s') {
for (var m of info['modes']) {
if (m['width'] == %(width)d &&
m['height'] == %(height)d) {
mode = m;
break;
}
}
break;
}
}
if (mode === undefined) {
console.error('Failed to select the resolution ' +
'%(width)dx%(height)d');
window.__set_resolution_progress = "mode not found";
return;
}
chrome.system.display.setDisplayProperties('%(id)s',
{'displayMode': mode}, () => {
if (chrome.runtime.lastError) {
window.__set_resolution_progress = "failed: " +
chrome.runtime.lastError.message;
} else {
window.__set_resolution_progress = "succeeded";
}
}
);
});
"""
% {'id': display_id, 'width': width, 'height': height}
)
utils.wait_for_value(lambda: (
extension.EvaluateJavaScript(
'window.__set_resolution_progress') != None),
expected_value=True)
result = extension.EvaluateJavaScript(
'window.__set_resolution_progress')
if result != 'succeeded':
raise RuntimeError('Failed to set resolution: %r' % result)
@_retry_display_call
def get_external_resolution(self):
"""Gets the resolution of the external screen.
@return The resolution tuple (width, height)
"""
return graphics_utils.get_external_resolution()
def get_internal_resolution(self):
"""Gets the resolution of the internal screen.
@return The resolution tuple (width, height) or None if internal screen
is not available
"""
for display in self.get_display_info():
if display['isInternal']:
bounds = display['bounds']
return (bounds['width'], bounds['height'])
return None
def set_content_protection(self, state):
"""Sets the content protection of the external screen.
@param state: One of the states 'Undesired', 'Desired', or 'Enabled'
"""
connector = self.get_external_connector_name()
graphics_utils.set_content_protection(connector, state)
def get_content_protection(self):
"""Gets the state of the content protection.
@param output: The output name as a string.
@return: A string of the state, like 'Undesired', 'Desired', or 'Enabled'.
False if not supported.
"""
connector = self.get_external_connector_name()
return graphics_utils.get_content_protection(connector)
def get_external_crtc_id(self):
"""Gets the external crtc.
@return The id of the external crtc."""
return graphics_utils.get_external_crtc_id()
def get_internal_crtc_id(self):
"""Gets the internal crtc.
@retrun The id of the internal crtc."""
return graphics_utils.get_internal_crtc_id()
def take_internal_screenshot(self, path):
"""Takes internal screenshot.
@param path: path to image file.
"""
self.take_screenshot_crtc(path, self.get_internal_crtc_id())
def take_external_screenshot(self, path):
"""Takes external screenshot.
@param path: path to image file.
"""
self.take_screenshot_crtc(path, self.get_external_crtc_id())
def take_screenshot_crtc(self, path, id):
"""Captures the DUT screenshot, use id for selecting screen.
@param path: path to image file.
@param id: The id of the crtc to screenshot.
"""
graphics_utils.take_screenshot_crop(path, crtc_id=id)
return True
def save_calibration_image(self, path):
"""Save the calibration image to the given path.
@param path: path to image file.
"""
shutil.copy(self.CALIBRATION_IMAGE_PATH, path)
return True
def take_tab_screenshot(self, output_path, url_pattern=None):
"""Takes a screenshot of the tab specified by the given url pattern.
@param output_path: A path of the output file.
@param url_pattern: A string of url pattern used to search for tabs.
Default is to look for .svg image.
"""
if url_pattern is None:
# If no URL pattern is provided, defaults to capture the first
# tab that shows SVG image.
url_pattern = '.svg'
tabs = self._resource.get_tabs()
for i in range(0, len(tabs)):
if url_pattern in tabs[i].url:
data = tabs[i].Screenshot(timeout=5)
# Flip the colors from BGR to RGB.
data = numpy.fliplr(data.reshape(-1, 3)).reshape(data.shape)
data.tofile(output_path)
break
return True
def toggle_mirrored(self):
"""Toggles mirrored."""
graphics_utils.screen_toggle_mirrored()
return True
def hide_cursor(self):
"""Hides mouse cursor."""
graphics_utils.hide_cursor()
return True
def hide_typing_cursor(self):
"""Hides typing cursor."""
graphics_utils.hide_typing_cursor()
return True
def is_mirrored_enabled(self):
"""Checks the mirrored state.
@return True if mirrored mode is enabled.
"""
return bool(self.get_display_info()[0]['mirroringSourceId'])
def set_mirrored(self, is_mirrored):
"""Sets mirrored mode.
@param is_mirrored: True or False to indicate mirrored state.
@return True if success, False otherwise.
"""
if self.is_mirrored_enabled() == is_mirrored:
return True
retries = 4
while retries > 0:
self.toggle_mirrored()
result = utils.wait_for_value(self.is_mirrored_enabled,
expected_value=is_mirrored,
timeout_sec=3)
if result == is_mirrored:
return True
retries -= 1
return False
def is_display_primary(self, internal=True):
"""Checks if internal screen is primary display.
@param internal: is internal/external screen primary status requested
@return boolean True if internal display is primary.
"""
for info in self.get_display_info():
if info['isInternal'] == internal and info['isPrimary']:
return True
return False
def suspend_resume(self, suspend_time=10):
"""Suspends the DUT for a given time in second.
@param suspend_time: Suspend time in second.
"""
sys_power.do_suspend(suspend_time)
return True
def suspend_resume_bg(self, suspend_time=10):
"""Suspends the DUT for a given time in second in the background.
@param suspend_time: Suspend time in second.
"""
process = multiprocessing.Process(target=self.suspend_resume,
args=(suspend_time,))
process.start()
return True
@_retry_display_call
def get_external_connector_name(self):
"""Gets the name of the external output connector.
@return The external output connector name as a string, if any.
Otherwise, return False.
"""
return graphics_utils.get_external_connector_name()
def get_internal_connector_name(self):
"""Gets the name of the internal output connector.
@return The internal output connector name as a string, if any.
Otherwise, return False.
"""
return graphics_utils.get_internal_connector_name()
def wait_external_display_connected(self, display):
"""Waits for the specified external display to be connected.
@param display: The display name as a string, like 'HDMI1', or
False if no external display is expected.
@return: True if display is connected; False otherwise.
"""
result = utils.wait_for_value(self.get_external_connector_name,
expected_value=display)
return result == display
@facade_resource.retry_chrome_call
def move_to_display(self, display_id):
"""Moves the current window to the indicated display.
@param display_id: The id of the indicated display.
@return True if success.
@raise TimeoutException if it fails.
"""
display_info = self._get_display_by_id(display_id)
if not display_info['isEnabled']:
raise RuntimeError('Cannot find the indicated display')
target_bounds = display_info['bounds']
extension = self._resource.get_extension()
# If the area of bounds is empty (here we achieve this by setting
# width and height to zero), the window_sizer will automatically
# determine an area which is visible and fits on the screen.
# For more details, see chrome/browser/ui/window_sizer.cc
# Without setting state to 'normal', if the current state is
# 'minimized', 'maximized' or 'fullscreen', the setting of
# 'left', 'top', 'width' and 'height' will be ignored.
# For more details, see chrome/browser/extensions/api/tabs/tabs_api.cc
extension.ExecuteJavaScript(
"""
var __status = 'Running';
chrome.windows.update(
chrome.windows.WINDOW_ID_CURRENT,
{left: %d, top: %d, width: 0, height: 0,
state: 'normal'},
function(info) {
if (info.left == %d && info.top == %d &&
info.state == 'normal')
__status = 'Done'; });
"""
% (target_bounds['left'], target_bounds['top'],
target_bounds['left'], target_bounds['top'])
)
extension.WaitForJavaScriptCondition(
"__status == 'Done'",
timeout=web_contents.DEFAULT_WEB_CONTENTS_TIMEOUT)
return True
def is_fullscreen_enabled(self):
"""Checks the fullscreen state.
@return True if fullscreen mode is enabled.
"""
return self.get_window_info()['state'] == 'fullscreen'
def set_fullscreen(self, is_fullscreen):
"""Sets the current window to full screen.
@param is_fullscreen: True or False to indicate fullscreen state.
@return True if success, False otherwise.
"""
extension = self._resource.get_extension()
if not extension:
raise RuntimeError('Autotest extension not found')
if is_fullscreen:
window_state = "fullscreen"
else:
window_state = "normal"
extension.ExecuteJavaScript(
"""
var __status = 'Running';
chrome.windows.update(
chrome.windows.WINDOW_ID_CURRENT,
{state: '%s'},
function() { __status = 'Done'; });
"""
% window_state)
utils.wait_for_value(lambda: (
extension.EvaluateJavaScript('__status') == 'Done'),
expected_value=True)
return self.is_fullscreen_enabled() == is_fullscreen
def load_url(self, url):
"""Loads the given url in a new tab. The new tab will be active.
@param url: The url to load as a string.
@return a str, the tab descriptor of the opened tab.
"""
return self._resource.load_url(url)
def load_calibration_image(self, resolution):
"""Opens a new tab and loads a full screen calibration
image from the HTTP server.
@param resolution: A tuple (width, height) of resolution.
@return a str, the tab descriptor of the opened tab.
"""
path = self.CALIBRATION_IMAGE_PATH
self._image_generator.generate_image(resolution[0], resolution[1], path)
os.chmod(path, 0o644)
tab_descriptor = self.load_url('file://%s' % path)
return tab_descriptor
def load_color_sequence(self, tab_descriptor, color_sequence):
"""Displays a series of colors on full screen on the tab.
tab_descriptor is returned by any open tab API of display facade.
e.g.,
tab_descriptor = load_url('about:blank')
load_color_sequence(tab_descriptor, color)
@param tab_descriptor: Indicate which tab to test.
@param color_sequence: An integer list for switching colors.
@return A list of the timestamp for each switch.
"""
tab = self._resource.get_tab_by_descriptor(tab_descriptor)
color_sequence_for_java_script = (
'var color_sequence = [' +
','.join("'#%06X'" % x for x in color_sequence) +
'];')
# Paints are synchronized to the fresh rate of the screen by
# window.requestAnimationFrame.
tab.ExecuteJavaScript(color_sequence_for_java_script + """
function render(timestamp) {
window.timestamp_list.push(timestamp);
if (window.count < color_sequence.length) {
document.body.style.backgroundColor =
color_sequence[count];
window.count++;
window.requestAnimationFrame(render);
}
}
window.count = 0;
window.timestamp_list = [];
window.requestAnimationFrame(render);
""")
# Waiting time is decided by following concerns:
# 1. MINIMUM_REFRESH_RATE_EXPECTED: the minimum refresh rate
# we expect it to be. Real refresh rate is related to
# not only hardware devices but also drivers and browsers.
# Most graphics devices support at least 60fps for a single
# monitor, and under mirror mode, since the both frames
# buffers need to be updated for an input frame, the refresh
# rate will decrease by half, so here we set it to be a
# little less than 30 (= 60/2) to make it more tolerant.
# 2. DELAY_TIME: extra wait time for timeout.
tab.WaitForJavaScriptCondition(
'window.count == color_sequence.length',
timeout=(
(len(color_sequence) / self.MINIMUM_REFRESH_RATE_EXPECTED)
+ self.DELAY_TIME))
return tab.EvaluateJavaScript("window.timestamp_list")
def close_tab(self, tab_descriptor):
"""Disables fullscreen and closes the tab of the given tab descriptor.
tab_descriptor is returned by any open tab API of display facade.
e.g.,
1.
tab_descriptor = load_url(url)
close_tab(tab_descriptor)
2.
tab_descriptor = load_calibration_image(resolution)
close_tab(tab_descriptor)
@param tab_descriptor: Indicate which tab to be closed.
"""
if tab_descriptor:
# set_fullscreen(False) is necessary here because currently there
# is a bug in tabs.Close(). If the current state is fullscreen and
# we call close_tab() without setting state back to normal, it will
# cancel fullscreen mode without changing system configuration, and
# so that the next time someone calls set_fullscreen(True), the
# function will find that current state is already 'fullscreen'
# (though it is not) and do nothing, which will break all the
# following tests.
self.set_fullscreen(False)
self._resource.close_tab(tab_descriptor)
else:
logging.error('close_tab: not a valid tab_descriptor')
return True
def reset_connector_if_applicable(self, connector_type):
"""Resets Type-C video connector from host end if applicable.
It's the workaround sequence since sometimes Type-C dongle becomes
corrupted and needs to be re-plugged.
@param connector_type: A string, like "VGA", "DVI", "HDMI", or "DP".
"""
if connector_type != 'HDMI' and connector_type != 'DP':
return
# Decide if we need to add --name=cros_pd
usbpd_command = 'ectool --name=cros_pd usbpd'
try:
common_utils.run('%s 0' % usbpd_command)
except error.CmdError:
usbpd_command = 'ectool usbpd'
port = 0
while port < self.MAX_TYPEC_PORT:
# We use usbpd to get Role information and then power cycle the
# SRC one.
command = '%s %d' % (usbpd_command, port)
try:
output = common_utils.run(command).stdout
if re.compile('Role.*SRC').search(output):
logging.info('power-cycle Type-C port %d', port)
common_utils.run('%s sink' % command)
common_utils.run('%s auto' % command)
port += 1
except error.CmdError:
break