| # Copyright (c) 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. |
| |
| import httplib |
| import logging |
| import os |
| import socket |
| import tempfile |
| import xmlrpclib |
| |
| from PIL import Image |
| |
| from autotest_lib.client.bin import utils |
| from autotest_lib.client.common_lib.cros import retry |
| from autotest_lib.client.cros import constants |
| from autotest_lib.server import autotest |
| from autotest_lib.server.cros.chameleon import image_generator |
| |
| |
| |
| class DisplayInfo(object): |
| """The class match displayInfo object from chrome.system.display API. |
| """ |
| |
| class Bounds(object): |
| """The class match Bounds object from chrome.system.display API. |
| |
| @param left: The x-coordinate of the upper-left corner. |
| @param top: The y-coordinate of the upper-left corner. |
| @param width: The width of the display in pixels. |
| @param height: The height of the display in pixels. |
| """ |
| def __init__(self, d): |
| self.left = d['left']; |
| self.top = d['top']; |
| self.width = d['width']; |
| self.height = d['height']; |
| |
| |
| class Insets(object): |
| """The class match Insets object from chrome.system.display API. |
| |
| @param left: The x-axis distance from the left bound. |
| @param left: The y-axis distance from the top bound. |
| @param left: The x-axis distance from the right bound. |
| @param left: The y-axis distance from the bottom bound. |
| """ |
| |
| def __init__(self, d): |
| self.left = d['left']; |
| self.top = d['top']; |
| self.right = d['right']; |
| self.bottom = d['bottom']; |
| |
| |
| def __init__(self, d): |
| self.display_id = d['id']; |
| self.name = d['name']; |
| self.mirroring_source_id = d['mirroringSourceId']; |
| self.is_primary = d['isPrimary']; |
| self.is_internal = d['isInternal']; |
| self.is_enabled = d['isEnabled']; |
| self.dpi_x = d['dpiX']; |
| self.dpi_y = d['dpiY']; |
| self.rotation = d['rotation']; |
| self.bounds = self.Bounds(d['bounds']); |
| self.overscan = self.Insets(d['overscan']); |
| self.work_area = self.Bounds(d['workArea']); |
| |
| |
| class DisplayClient(object): |
| """DisplayClient is a layer to control display logic over a remote DUT. |
| |
| The Autotest host object representing the remote DUT, passed to this |
| class on initialization, can be accessed from its _client property. |
| |
| """ |
| |
| X_ENV_VARIABLES = 'DISPLAY=:0.0 XAUTHORITY=/home/chronos/.Xauthority' |
| XMLRPC_CONNECT_TIMEOUT = 60 |
| XMLRPC_RETRY_TIMEOUT = 180 |
| XMLRPC_RETRY_DELAY = 10 |
| HTTP_PORT = 8000 |
| DEST_TMP_DIR = '/tmp' |
| DEST_IMAGE_FILENAME = 'calibration.svg' |
| |
| |
| def __init__(self, host): |
| """Construct a DisplayClient. |
| |
| @param host: Host object representing a remote host. |
| """ |
| self._client = host |
| self._display_xmlrpc_client = None |
| self._image_generator = image_generator.ImageGenerator() |
| |
| |
| def initialize(self): |
| """Initializes some required servers, like HTTP daemon, RPC connection. |
| """ |
| # Make sure the client library is on the device so that the proxy code |
| # is there when we try to call it. |
| client_at = autotest.Autotest(self._client) |
| client_at.install() |
| self.connect() |
| |
| |
| def connect(self): |
| """Connects the XML-RPC proxy on the client.""" |
| @retry.retry((socket.error, |
| xmlrpclib.ProtocolError, |
| httplib.BadStatusLine), |
| timeout_min=self.XMLRPC_RETRY_TIMEOUT / 60.0, |
| delay_sec=self.XMLRPC_RETRY_DELAY) |
| def connect_with_retries(): |
| """Connects the XML-RPC proxy with retries.""" |
| self._display_xmlrpc_client = self._client.xmlrpc_connect( |
| constants.DISPLAY_TESTING_XMLRPC_SERVER_COMMAND, |
| constants.DISPLAY_TESTING_XMLRPC_SERVER_PORT, |
| command_name=( |
| constants.DISPLAY_TESTING_XMLRPC_SERVER_CLEANUP_PATTERN |
| ), |
| ready_test_name=( |
| constants.DISPLAY_TESTING_XMLRPC_SERVER_READY_METHOD), |
| timeout_seconds=self.XMLRPC_CONNECT_TIMEOUT) |
| |
| logging.info('Setup the display_client RPC server, with retries...') |
| connect_with_retries() |
| |
| |
| def cleanup(self): |
| """Cleans up.""" |
| self._client.rpc_disconnect( |
| constants.DISPLAY_TESTING_XMLRPC_SERVER_PORT) |
| |
| |
| def __del__(self): |
| """Destructor of DisplayClient.""" |
| self.cleanup() |
| |
| |
| def get_connector_name(self): |
| """Gets the name of the external output connector. |
| |
| @return The external output connector name as a string. |
| """ |
| return self._display_xmlrpc_client.get_ext_connector_name() |
| |
| |
| def load_calibration_image(self, resolution): |
| """Load a full screen calibration image from the HTTP server. |
| |
| @param resolution: A tuple (width, height) of resolution. |
| """ |
| with tempfile.NamedTemporaryFile() as f: |
| self._image_generator.generate_image( |
| resolution[0], resolution[1], f.name) |
| os.chmod(f.name, 0644) |
| self._client.send_file( |
| f.name, |
| os.path.join(self.DEST_TMP_DIR, self.DEST_IMAGE_FILENAME)) |
| |
| page_url = 'file://%s/%s' % (self.DEST_TMP_DIR, |
| self.DEST_IMAGE_FILENAME) |
| self._display_xmlrpc_client.load_url(page_url) |
| |
| |
| def close_tab(self, index=-1): |
| """Closes the tab of the given index. |
| |
| @param index: The tab index to close. Defaults to the last tab. |
| """ |
| return self._display_xmlrpc_client.close_tab(index) |
| |
| |
| def set_mirrored(self, is_mirrored): |
| """Sets mirrored mode. |
| |
| @param is_mirrored: True or False to indicate mirrored state. |
| """ |
| return self._display_xmlrpc_client.set_mirrored(is_mirrored) |
| |
| |
| def suspend_resume(self, suspend_time=10): |
| """Suspends the DUT for a given time in second. |
| |
| @param suspend_time: Suspend time in second, default: 10s. |
| """ |
| # TODO(waihong): Use other general API instead of this RPC. |
| return self._display_xmlrpc_client.suspend_resume(suspend_time) |
| |
| |
| 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, default: 10s. |
| """ |
| # TODO(waihong): Use other general API instead of this RPC. |
| return self._display_xmlrpc_client.suspend_resume_bg(suspend_time) |
| |
| |
| def reconnect_output_and_wait(self, reconnect=True, |
| expected_display_count=2): |
| """Reconnects output and waits it available. |
| |
| @param reconnect: True to perform a re-connection from the DUT; False |
| otherwise. |
| @param expected_display_count: |
| number of displays expected to be connected. |
| """ |
| output = self.get_connector_name() |
| if reconnect: |
| self._display_xmlrpc_client.reconnect_output(output) |
| self._display_xmlrpc_client.wait_output_connected(output) |
| utils.wait_for_value(lambda: ( |
| len(self._display_xmlrpc_client.get_display_info())), |
| expected_value=expected_display_count) |
| |
| |
| def hide_cursor(self): |
| """Hides mouse cursor by sending a keystroke.""" |
| self._display_xmlrpc_client.press_key('Up') |
| |
| |
| def capture_external_screen(self): |
| """Captures the external screen framebuffer. |
| |
| @return: An Image object. |
| """ |
| output = self.get_connector_name() |
| fb_w, fb_h, fb_x, fb_y = ( |
| self._display_xmlrpc_client.get_resolution(output)) |
| with tempfile.NamedTemporaryFile(suffix='.rgb') as f: |
| basename = os.path.basename(f.name) |
| remote_path = os.path.join('/tmp', basename) |
| command = ('%s import -window root -depth 8 -crop %dx%d+%d+%d %s' % |
| (self.X_ENV_VARIABLES, fb_w, fb_h, fb_x, fb_y, |
| remote_path)) |
| self._client.run(command) |
| self._client.get_file(remote_path, f.name) |
| return Image.fromstring('RGB', (fb_w, fb_h), open(f.name).read()) |
| |
| |
| def get_resolution(self): |
| """Gets the external resolution on framebuffer. |
| |
| @return The resolution tuple (width, height) |
| """ |
| output = self.get_connector_name() |
| width, height, _, _ = self._display_xmlrpc_client.get_resolution(output) |
| return (width, height) |
| |
| |
| def set_resolution(self, display_index, width, height): |
| """Sets the resolution on the specified display. |
| |
| @param display_index: index of the display to set resolutions for; 0 is |
| the internal one for chromebooks. |
| @param width: width of the resolution |
| @param height: height of the resolution |
| """ |
| self._display_xmlrpc_client.set_resolution( |
| display_index, width, height) |
| |
| |
| def get_display_info(self): |
| """Gets the information of all the displays that are connected to the |
| DUT. |
| |
| @return: array of object DisplayInfo for display informtion |
| """ |
| return map(DisplayInfo, self._display_xmlrpc_client.get_display_info()) |
| |
| |
| def get_available_resolutions(self, display_index): |
| """Gets the available hardware resolutions of the specified display. |
| |
| @param display_index: index of the display to get resolutions from; the |
| index is from the DisplayInfo array obtained by get_display_info(). |
| For Chromebooks, index 0 indicates the internal display. |
| |
| @return: array of available hardware resolutions tuple (width, height) |
| """ |
| # Previous version before CR:417113012 (targeted for M38) |
| # (resolution['width'],resolution['height']) |
| |
| # TODO(tingyuan): fix loading test image for cases where original |
| # width/height is different from width/height. |
| return list(set([(resolution['originalWidth'], |
| resolution['originalHeight']) |
| for resolution in |
| self._display_xmlrpc_client.get_available_resolutions( |
| display_index)])) |
