| # Lint as: python2, python3 |
| # 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. |
| |
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
| import atexit |
| import six.moves.http_client |
| import logging |
| import os |
| import socket |
| import time |
| from six.moves import range |
| import six.moves.xmlrpc_client |
| from contextlib import contextmanager |
| |
| try: |
| from PIL import Image |
| except ImportError: |
| Image = None |
| |
| from autotest_lib.client.bin import utils |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.cros.chameleon import audio_board |
| from autotest_lib.client.cros.chameleon import chameleon_bluetooth_audio |
| from autotest_lib.client.cros.chameleon import edid as edid_lib |
| from autotest_lib.client.cros.chameleon import usb_controller |
| |
| |
| CHAMELEON_PORT = 9992 |
| CHAMELEOND_LOG_REMOTE_PATH = '/var/log/chameleond' |
| DAEMON_LOG_REMOTE_PATH = '/var/log/daemon.log' |
| BTMON_LOG_REMOTE_PATH = '/var/log/btsnoop.log' |
| CHAMELEON_READY_TEST = 'GetSupportedPorts' |
| |
| |
| class ChameleonConnectionError(error.TestError): |
| """Indicates that connecting to Chameleon failed. |
| |
| It is fatal to the test unless caught. |
| """ |
| pass |
| |
| |
| class _Method(object): |
| """Class to save the name of the RPC method instead of the real object. |
| |
| It keeps the name of the RPC method locally first such that the RPC method |
| can be evaluated to a real object while it is called. Its purpose is to |
| refer to the latest RPC proxy as the original previous-saved RPC proxy may |
| be lost due to reboot. |
| |
| The call_server is the method which does refer to the latest RPC proxy. |
| |
| This class and the re-connection mechanism in ChameleonConnection is |
| copied from third_party/autotest/files/server/cros/faft/rpc_proxy.py |
| |
| """ |
| def __init__(self, call_server, name): |
| """Constructs a _Method. |
| |
| @param call_server: the call_server method |
| @param name: the method name or instance name provided by the |
| remote server |
| |
| """ |
| self.__call_server = call_server |
| self._name = name |
| |
| |
| def __getattr__(self, name): |
| """Support a nested method. |
| |
| For example, proxy.system.listMethods() would need to use this method |
| to get system and then to get listMethods. |
| |
| @param name: the method name or instance name provided by the |
| remote server |
| |
| @return: a callable _Method object. |
| |
| """ |
| return _Method(self.__call_server, "%s.%s" % (self._name, name)) |
| |
| |
| def __call__(self, *args, **dargs): |
| """The call method of the object. |
| |
| @param args: arguments for the remote method. |
| @param kwargs: keyword arguments for the remote method. |
| |
| @return: the result returned by the remote method. |
| |
| """ |
| return self.__call_server(self._name, *args, **dargs) |
| |
| |
| class ChameleonConnection(object): |
| """ChameleonConnection abstracts the network connection to the board. |
| |
| When a chameleon board is rebooted, a xmlrpc call would incur a |
| socket error. To fix the error, a client has to reconnect to the server. |
| ChameleonConnection is a wrapper of chameleond proxy created by |
| xmlrpclib.ServerProxy(). ChameleonConnection has the capability to |
| automatically reconnect to the server when such socket error occurs. |
| The nice feature is that the auto re-connection is performed inside this |
| wrapper and is transparent to the caller. |
| |
| Note: |
| 1. When running chameleon autotests in lab machines, it is |
| ChameleonConnection._create_server_proxy() that is invoked. |
| 2. When running chameleon autotests in local chroot, it is |
| rpc_server_tracker.xmlrpc_connect() in server/hosts/chameleon_host.py |
| that is invoked. |
| |
| ChameleonBoard and ChameleonPort use it for accessing Chameleon RPC. |
| |
| """ |
| |
| def __init__(self, hostname, port=CHAMELEON_PORT, proxy_generator=None, |
| ready_test_name=CHAMELEON_READY_TEST): |
| """Constructs a ChameleonConnection. |
| |
| @param hostname: Hostname the chameleond process is running. |
| @param port: Port number the chameleond process is listening on. |
| @param proxy_generator: a function to generate server proxy. |
| @param ready_test_name: run this method on the remote server ot test |
| if the server is connected correctly. |
| |
| @raise ChameleonConnectionError if connection failed. |
| """ |
| self._hostname = hostname |
| self._port = port |
| |
| # Note: it is difficult to put the lambda function as the default |
| # value of the proxy_generator argument. In that case, the binding |
| # of arguments (hostname and port) would be delayed until run time |
| # which requires to pass an instance as an argument to labmda. |
| # That becomes cumbersome since server/hosts/chameleon_host.py |
| # would also pass a lambda without argument to instantiate this object. |
| # Use the labmda function as follows would bind the needed arguments |
| # immediately which is much simpler. |
| self._proxy_generator = proxy_generator or self._create_server_proxy |
| |
| self._ready_test_name = ready_test_name |
| self._chameleond_proxy = None |
| |
| |
| def _create_server_proxy(self): |
| """Creates the chameleond server proxy. |
| |
| @param hostname: Hostname the chameleond process is running. |
| @param port: Port number the chameleond process is listening on. |
| |
| @return ServerProxy object to chameleond. |
| |
| @raise ChameleonConnectionError if connection failed. |
| |
| """ |
| remote = 'http://%s:%s' % (self._hostname, self._port) |
| chameleond_proxy = six.moves.xmlrpc_client.ServerProxy(remote, allow_none=True) |
| logging.info('ChameleonConnection._create_server_proxy() called') |
| # Call a RPC to test. |
| try: |
| getattr(chameleond_proxy, self._ready_test_name)() |
| except (socket.error, |
| six.moves.xmlrpc_client.ProtocolError, |
| six.moves.http_client.BadStatusLine) as e: |
| raise ChameleonConnectionError(e) |
| return chameleond_proxy |
| |
| |
| def _reconnect(self): |
| """Reconnect to chameleond.""" |
| self._chameleond_proxy = self._proxy_generator() |
| |
| |
| def __call_server(self, name, *args, **kwargs): |
| """Bind the name to the chameleond proxy and execute the method. |
| |
| @param name: the method name or instance name provided by the |
| remote server. |
| @param args: arguments for the remote method. |
| @param kwargs: keyword arguments for the remote method. |
| |
| @return: the result returned by the remote method. |
| |
| @raise ChameleonConnectionError if the call failed after a reconnection. |
| |
| """ |
| try: |
| return getattr(self._chameleond_proxy, name)(*args, **kwargs) |
| except (AttributeError, socket.error): |
| # Reconnect and invoke the method again. |
| logging.info('Reconnecting chameleond proxy: %s', name) |
| self._reconnect() |
| try: |
| return getattr(self._chameleond_proxy, name)(*args, **kwargs) |
| except (socket.error) as e: |
| raise ChameleonConnectionError( |
| ("The RPC call %s still failed with %s" |
| " after a reconnection.") % (name, e)) |
| return None |
| |
| def __getattr__(self, name): |
| """Get the callable _Method object. |
| |
| @param name: the method name or instance name provided by the |
| remote server |
| |
| @return: a callable _Method object. |
| |
| """ |
| return _Method(self.__call_server, name) |
| |
| |
| class ChameleonBoard(object): |
| """ChameleonBoard is an abstraction of a Chameleon board. |
| |
| A Chameleond RPC proxy is passed to the construction such that it can |
| use this proxy to control the Chameleon board. |
| |
| User can use host to access utilities that are not provided by |
| Chameleond XMLRPC server, e.g. send_file and get_file, which are provided by |
| ssh_host.SSHHost, which is the base class of ChameleonHost. |
| |
| """ |
| |
| def __init__(self, chameleon_connection, chameleon_host=None): |
| """Construct a ChameleonBoard. |
| |
| @param chameleon_connection: ChameleonConnection object. |
| @param chameleon_host: ChameleonHost object. None if this ChameleonBoard |
| is not created by a ChameleonHost. |
| """ |
| self.host = chameleon_host |
| self._output_log_file = None |
| self._chameleond_proxy = chameleon_connection |
| self._usb_ctrl = usb_controller.USBController(chameleon_connection) |
| if self._chameleond_proxy.HasAudioBoard(): |
| self._audio_board = audio_board.AudioBoard(chameleon_connection) |
| else: |
| self._audio_board = None |
| logging.info('There is no audio board on this Chameleon.') |
| self._bluetooth_ref_controller = ( |
| chameleon_bluetooth_audio. |
| BluetoothRefController(chameleon_connection) |
| ) |
| |
| |
| def reset(self): |
| """Resets Chameleon board.""" |
| self._chameleond_proxy.Reset() |
| |
| |
| def setup_and_reset(self, output_dir=None): |
| """Setup and reset Chameleon board. |
| |
| @param output_dir: Setup the output directory. |
| None for just reset the board. |
| """ |
| if output_dir and self.host is not None: |
| logging.info('setup_and_reset: dir %s, chameleon host %s', |
| output_dir, self.host.hostname) |
| log_dir = os.path.join(output_dir, 'chameleond', self.host.hostname) |
| # Only clear the chameleon board log and register get log callback |
| # when we first create the log_dir. |
| if not os.path.exists(log_dir): |
| # remove old log. |
| self.host.run('>%s' % CHAMELEOND_LOG_REMOTE_PATH) |
| os.makedirs(log_dir) |
| self._output_log_file = os.path.join(log_dir, 'log') |
| atexit.register(self._get_log) |
| self.reset() |
| |
| |
| def register_raspPi_log(self, output_dir): |
| """Register log for raspberry Pi |
| |
| This method log bluetooth related files on Raspberry Pi. |
| If the host is not running on Raspberry Pi, some files may be ignored. |
| """ |
| log_dir = os.path.join(output_dir, 'chameleond', self.host.hostname) |
| |
| if not os.path.exists(log_dir): |
| os.makedirs(log_dir) |
| |
| def log_new_gen(source_path): |
| """Generate function to save logs logging during the test |
| |
| @param source_path: The log file path that want to be saved |
| |
| @return: Function to save the logs if file in source_path exists, |
| None otherwise. |
| """ |
| |
| # Check if the file exists |
| file_exist = self.host.run('[ -f %s ] || echo "not found"' % |
| source_path).stdout.strip() |
| if file_exist == 'not found': |
| return None |
| |
| byte_to_skip = self.host.run('stat --printf="%%s" %s' % |
| source_path).stdout.strip() |
| file_name = os.path.basename(source_path) |
| target_path = os.path.join(log_dir, file_name) |
| |
| def log_new(): |
| """Save the newly added logs""" |
| tmp_file_path = source_path+'.new' |
| |
| # Store a temporary file with newly added content |
| # Set the start point as byte_to_skip + 1 |
| self.host.run('tail -c +%s %s > %s' % (int(byte_to_skip)+1, |
| source_path, |
| tmp_file_path)) |
| self.host.get_file(tmp_file_path, target_path) |
| self.host.run('rm %s' % tmp_file_path) |
| return log_new |
| |
| for source_path in [CHAMELEOND_LOG_REMOTE_PATH, DAEMON_LOG_REMOTE_PATH]: |
| log_new_func = log_new_gen(source_path) |
| if log_new_func: |
| atexit.register(log_new_func) |
| |
| |
| def btmon_atexit_gen(btmon_pid): |
| """Generate a function to kill the btmon process and save the log |
| |
| @param btmon_pid: PID of the btmon process |
| """ |
| |
| def btmon_atexit(): |
| """Kill the btmon with specified PID and save the log""" |
| |
| file_name = os.path.basename(BTMON_LOG_REMOTE_PATH) |
| target_path = os.path.join(log_dir, file_name) |
| |
| self.host.run('kill %d' % btmon_pid) |
| self.host.get_file(BTMON_LOG_REMOTE_PATH, target_path) |
| return btmon_atexit |
| |
| |
| # Kill all btmon process before creating a new one |
| self.host.run('pkill btmon || true') |
| |
| # Get available btmon options in the chameleon host |
| btmon_options = '' |
| btmon_help = self.host.run('btmon --help').stdout |
| |
| for option in 'SA': |
| if '-%s' % option in btmon_help: |
| btmon_options += option |
| |
| # Store btmon log |
| btmon_pid = int(self.host.run_background('btmon -%sw %s' |
| % (btmon_options, |
| BTMON_LOG_REMOTE_PATH))) |
| if btmon_pid > 0: |
| atexit.register(btmon_atexit_gen(btmon_pid)) |
| |
| |
| def reboot(self): |
| """Reboots Chameleon board.""" |
| self._chameleond_proxy.Reboot() |
| |
| |
| def get_bt_commit_hash(self): |
| """ Read the current git commit hash of chameleond.""" |
| return self._chameleond_proxy.get_bt_commit_hash() |
| |
| |
| def _get_log(self): |
| """Get log from chameleon. It will be registered by atexit. |
| |
| It's a private method. We will setup output_dir before using this |
| method. |
| """ |
| self.host.get_file(CHAMELEOND_LOG_REMOTE_PATH, self._output_log_file) |
| |
| def log_message(self, msg): |
| """Log a message in chameleond log and system log.""" |
| self._chameleond_proxy.log_message(msg) |
| |
| def get_all_ports(self): |
| """Gets all the ports on Chameleon board which are connected. |
| |
| @return: A list of ChameleonPort objects. |
| """ |
| ports = self._chameleond_proxy.ProbePorts() |
| return [ChameleonPort(self._chameleond_proxy, port) for port in ports] |
| |
| |
| def get_all_inputs(self): |
| """Gets all the input ports on Chameleon board which are connected. |
| |
| @return: A list of ChameleonPort objects. |
| """ |
| ports = self._chameleond_proxy.ProbeInputs() |
| return [ChameleonPort(self._chameleond_proxy, port) for port in ports] |
| |
| |
| def get_all_outputs(self): |
| """Gets all the output ports on Chameleon board which are connected. |
| |
| @return: A list of ChameleonPort objects. |
| """ |
| ports = self._chameleond_proxy.ProbeOutputs() |
| return [ChameleonPort(self._chameleond_proxy, port) for port in ports] |
| |
| |
| def get_label(self): |
| """Gets the label which indicates the display connection. |
| |
| @return: A string of the label, like 'hdmi', 'dp_hdmi', etc. |
| """ |
| connectors = [] |
| for port in self._chameleond_proxy.ProbeInputs(): |
| if self._chameleond_proxy.HasVideoSupport(port): |
| connector = self._chameleond_proxy.GetConnectorType(port).lower() |
| connectors.append(connector) |
| # Eliminate duplicated ports. It simplifies the labels of dual-port |
| # devices, i.e. dp_dp categorized into dp. |
| return '_'.join(sorted(set(connectors))) |
| |
| |
| def get_audio_board(self): |
| """Gets the audio board on Chameleon. |
| |
| @return: An AudioBoard object. |
| """ |
| return self._audio_board |
| |
| |
| def get_usb_controller(self): |
| """Gets the USB controller on Chameleon. |
| |
| @return: A USBController object. |
| """ |
| return self._usb_ctrl |
| |
| |
| def get_bluetooth_base(self): |
| """Gets the Bluetooth base object on Chameleon. |
| |
| This is a base object that does not emulate any Bluetooth device. |
| |
| @return: A BluetoothBaseFlow object. |
| """ |
| return self._chameleond_proxy.bluetooth_base |
| |
| |
| def get_bluetooth_tester(self): |
| """Gets the Bluetooth tester object on Chameleon. |
| |
| @return: A BluetoothTester object. |
| """ |
| return self._chameleond_proxy.bluetooth_tester |
| |
| |
| def get_bluetooth_audio(self): |
| """Gets the Bluetooth audio object on Chameleon. |
| |
| @return: A RaspiBluetoothAudioFlow object. |
| """ |
| return self._chameleond_proxy.bluetooth_audio |
| |
| |
| def get_bluetooth_hid_mouse(self): |
| """Gets the emulated Bluetooth (BR/EDR) HID mouse on Chameleon. |
| |
| @return: A BluetoothHIDMouseFlow object. |
| """ |
| return self._chameleond_proxy.bluetooth_mouse |
| |
| |
| def get_bluetooth_hid_keyboard(self): |
| """Gets the emulated Bluetooth (BR/EDR) HID keyboard on Chameleon. |
| |
| @return: A BluetoothHIDKeyboardFlow object. |
| """ |
| return self._chameleond_proxy.bluetooth_keyboard |
| |
| |
| def get_bluetooth_ref_controller(self): |
| """Gets the emulated BluetoothRefController. |
| |
| @return: A BluetoothRefController object. |
| """ |
| return self._bluetooth_ref_controller |
| |
| |
| def get_avsync_probe(self): |
| """Gets the avsync probe device on Chameleon. |
| |
| @return: An AVSyncProbeFlow object. |
| """ |
| return self._chameleond_proxy.avsync_probe |
| |
| |
| def get_motor_board(self): |
| """Gets the motor_board device on Chameleon. |
| |
| @return: An MotorBoard object. |
| """ |
| return self._chameleond_proxy.motor_board |
| |
| |
| def get_usb_printer(self): |
| """Gets the printer device on Chameleon. |
| |
| @return: A printer object. |
| """ |
| return self._chameleond_proxy.printer |
| |
| |
| def get_mac_address(self): |
| """Gets the MAC address of Chameleon. |
| |
| @return: A string for MAC address. |
| """ |
| return self._chameleond_proxy.GetMacAddress() |
| |
| |
| def get_bluetooth_a2dp_sink(self): |
| """Gets the Bluetooth A2DP sink on chameleon host. |
| |
| @return: A BluetoothA2DPSinkFlow object. |
| """ |
| return self._chameleond_proxy.bluetooth_a2dp_sink |
| |
| def get_ble_mouse(self): |
| """Gets the BLE mouse (nRF52) on chameleon host. |
| |
| @return: A BluetoothHIDFlow object. |
| """ |
| return self._chameleond_proxy.ble_mouse |
| |
| def get_ble_keyboard(self): |
| """Gets the BLE keyboard on chameleon host. |
| |
| @return: A BluetoothHIDFlow object. |
| """ |
| return self._chameleond_proxy.ble_keyboard |
| |
| def get_ble_phone(self): |
| """Gets the emulated Bluetooth phone on Chameleon. |
| |
| @return: A RaspiPhone object. |
| """ |
| return self._chameleond_proxy.ble_phone |
| |
| def get_platform(self): |
| """ Get the Hardware Platform of the chameleon host |
| |
| @return: CHROMEOS/RASPI |
| """ |
| return self._chameleond_proxy.get_platform() |
| |
| |
| class ChameleonPort(object): |
| """ChameleonPort is an abstraction of a general port of a Chameleon board. |
| |
| It only contains some common methods shared with audio and video ports. |
| |
| A Chameleond RPC proxy and an port_id are passed to the construction. |
| The port_id is the unique identity to the port. |
| """ |
| |
| def __init__(self, chameleond_proxy, port_id): |
| """Construct a ChameleonPort. |
| |
| @param chameleond_proxy: Chameleond RPC proxy object. |
| @param port_id: The ID of the input port. |
| """ |
| self.chameleond_proxy = chameleond_proxy |
| self.port_id = port_id |
| |
| |
| def get_connector_id(self): |
| """Returns the connector ID. |
| |
| @return: A number of connector ID. |
| """ |
| return self.port_id |
| |
| |
| def get_connector_type(self): |
| """Returns the human readable string for the connector type. |
| |
| @return: A string, like "VGA", "DVI", "HDMI", or "DP". |
| """ |
| return self.chameleond_proxy.GetConnectorType(self.port_id) |
| |
| |
| def has_audio_support(self): |
| """Returns if the input has audio support. |
| |
| @return: True if the input has audio support; otherwise, False. |
| """ |
| return self.chameleond_proxy.HasAudioSupport(self.port_id) |
| |
| |
| def has_video_support(self): |
| """Returns if the input has video support. |
| |
| @return: True if the input has video support; otherwise, False. |
| """ |
| return self.chameleond_proxy.HasVideoSupport(self.port_id) |
| |
| |
| def plug(self): |
| """Asserts HPD line to high, emulating plug.""" |
| logging.info('Plug Chameleon port %d', self.port_id) |
| self.chameleond_proxy.Plug(self.port_id) |
| |
| |
| def unplug(self): |
| """Deasserts HPD line to low, emulating unplug.""" |
| logging.info('Unplug Chameleon port %d', self.port_id) |
| self.chameleond_proxy.Unplug(self.port_id) |
| |
| |
| def set_plug(self, plug_status): |
| """Sets plug/unplug by plug_status. |
| |
| @param plug_status: True to plug; False to unplug. |
| """ |
| if plug_status: |
| self.plug() |
| else: |
| self.unplug() |
| |
| |
| @property |
| def plugged(self): |
| """ |
| @returns True if this port is plugged to Chameleon, False otherwise. |
| |
| """ |
| return self.chameleond_proxy.IsPlugged(self.port_id) |
| |
| |
| class ChameleonVideoInput(ChameleonPort): |
| """ChameleonVideoInput is an abstraction of a video input port. |
| |
| It contains some special methods to control a video input. |
| """ |
| |
| _DUT_STABILIZE_TIME = 3 |
| _DURATION_UNPLUG_FOR_EDID = 5 |
| _TIMEOUT_VIDEO_STABLE_PROBE = 10 |
| _EDID_ID_DISABLE = -1 |
| _FRAME_RATE = 60 |
| |
| def __init__(self, chameleon_port): |
| """Construct a ChameleonVideoInput. |
| |
| @param chameleon_port: A general ChameleonPort object. |
| """ |
| self.chameleond_proxy = chameleon_port.chameleond_proxy |
| self.port_id = chameleon_port.port_id |
| self._original_edid = None |
| |
| |
| def wait_video_input_stable(self, timeout=None): |
| """Waits the video input stable or timeout. |
| |
| @param timeout: The time period to wait for. |
| |
| @return: True if the video input becomes stable within the timeout |
| period; otherwise, False. |
| """ |
| is_input_stable = self.chameleond_proxy.WaitVideoInputStable( |
| self.port_id, timeout) |
| |
| # If video input of Chameleon has been stable, wait for DUT software |
| # layer to be stable as well to make sure all the configurations have |
| # been propagated before proceeding. |
| if is_input_stable: |
| logging.info('Video input has been stable. Waiting for the DUT' |
| ' to be stable...') |
| time.sleep(self._DUT_STABILIZE_TIME) |
| return is_input_stable |
| |
| |
| def read_edid(self): |
| """Reads the EDID. |
| |
| @return: An Edid object or NO_EDID. |
| """ |
| edid_binary = self.chameleond_proxy.ReadEdid(self.port_id) |
| if edid_binary is None: |
| return edid_lib.NO_EDID |
| # Read EDID without verify. It may be made corrupted as intended |
| # for the test purpose. |
| return edid_lib.Edid(edid_binary.data, skip_verify=True) |
| |
| |
| def apply_edid(self, edid): |
| """Applies the given EDID. |
| |
| @param edid: An Edid object or NO_EDID. |
| """ |
| if edid is edid_lib.NO_EDID: |
| self.chameleond_proxy.ApplyEdid(self.port_id, self._EDID_ID_DISABLE) |
| else: |
| edid_binary = six.moves.xmlrpc_client.Binary(edid.data) |
| edid_id = self.chameleond_proxy.CreateEdid(edid_binary) |
| self.chameleond_proxy.ApplyEdid(self.port_id, edid_id) |
| self.chameleond_proxy.DestroyEdid(edid_id) |
| |
| def set_edid_from_file(self, filename, check_video_input=True): |
| """Sets EDID from a file. |
| |
| The method is similar to set_edid but reads EDID from a file. |
| |
| @param filename: path to EDID file. |
| @param check_video_input: False to disable wait_video_input_stable. |
| """ |
| self.set_edid(edid_lib.Edid.from_file(filename), |
| check_video_input=check_video_input) |
| |
| def set_edid(self, edid, check_video_input=True): |
| """The complete flow of setting EDID. |
| |
| Unplugs the port if needed, sets EDID, plugs back if it was plugged. |
| The original EDID is stored so user can call restore_edid after this |
| call. |
| |
| @param edid: An Edid object. |
| @param check_video_input: False to disable wait_video_input_stable. |
| """ |
| plugged = self.plugged |
| if plugged: |
| self.unplug() |
| |
| self._original_edid = self.read_edid() |
| |
| logging.info('Apply EDID on port %d', self.port_id) |
| self.apply_edid(edid) |
| |
| if plugged: |
| time.sleep(self._DURATION_UNPLUG_FOR_EDID) |
| self.plug() |
| if check_video_input: |
| self.wait_video_input_stable(self._TIMEOUT_VIDEO_STABLE_PROBE) |
| |
| def restore_edid(self): |
| """Restores original EDID stored when set_edid was called.""" |
| current_edid = self.read_edid() |
| if (self._original_edid and |
| self._original_edid.data != current_edid.data): |
| logging.info('Restore the original EDID.') |
| self.apply_edid(self._original_edid) |
| |
| |
| @contextmanager |
| def use_edid(self, edid, check_video_input=True): |
| """Uses the given EDID in a with statement. |
| |
| It sets the EDID up in the beginning and restores to the original |
| EDID in the end. This function is expected to be used in a with |
| statement, like the following: |
| |
| with chameleon_port.use_edid(edid): |
| do_some_test_on(chameleon_port) |
| |
| @param edid: An EDID object. |
| @param check_video_input: False to disable wait_video_input_stable. |
| """ |
| # Set the EDID up in the beginning. |
| self.set_edid(edid, check_video_input=check_video_input) |
| |
| try: |
| # Yeild to execute the with statement. |
| yield |
| finally: |
| # Restore the original EDID in the end. |
| self.restore_edid() |
| |
| def use_edid_file(self, filename, check_video_input=True): |
| """Uses the given EDID file in a with statement. |
| |
| It sets the EDID up in the beginning and restores to the original |
| EDID in the end. This function is expected to be used in a with |
| statement, like the following: |
| |
| with chameleon_port.use_edid_file(filename): |
| do_some_test_on(chameleon_port) |
| |
| @param filename: A path to the EDID file. |
| @param check_video_input: False to disable wait_video_input_stable. |
| """ |
| return self.use_edid(edid_lib.Edid.from_file(filename), |
| check_video_input=check_video_input) |
| |
| def fire_hpd_pulse(self, deassert_interval_usec, assert_interval_usec=None, |
| repeat_count=1, end_level=1): |
| |
| """Fires one or more HPD pulse (low -> high -> low -> ...). |
| |
| @param deassert_interval_usec: The time in microsecond of the |
| deassert pulse. |
| @param assert_interval_usec: The time in microsecond of the |
| assert pulse. If None, then use the same value as |
| deassert_interval_usec. |
| @param repeat_count: The count of HPD pulses to fire. |
| @param end_level: HPD ends with 0 for LOW (unplugged) or 1 for |
| HIGH (plugged). |
| """ |
| self.chameleond_proxy.FireHpdPulse( |
| self.port_id, deassert_interval_usec, |
| assert_interval_usec, repeat_count, int(bool(end_level))) |
| |
| |
| def fire_mixed_hpd_pulses(self, widths): |
| """Fires one or more HPD pulses, starting at low, of mixed widths. |
| |
| One must specify a list of segment widths in the widths argument where |
| widths[0] is the width of the first low segment, widths[1] is that of |
| the first high segment, widths[2] is that of the second low segment... |
| etc. The HPD line stops at low if even number of segment widths are |
| specified; otherwise, it stops at high. |
| |
| @param widths: list of pulse segment widths in usec. |
| """ |
| self.chameleond_proxy.FireMixedHpdPulses(self.port_id, widths) |
| |
| |
| def capture_screen(self): |
| """Captures Chameleon framebuffer. |
| |
| @return An Image object. |
| """ |
| return Image.fromstring( |
| 'RGB', |
| self.get_resolution(), |
| self.chameleond_proxy.DumpPixels(self.port_id).data) |
| |
| |
| def get_resolution(self): |
| """Gets the source resolution. |
| |
| @return: A (width, height) tuple. |
| """ |
| # The return value of RPC is converted to a list. Convert it back to |
| # a tuple. |
| return tuple(self.chameleond_proxy.DetectResolution(self.port_id)) |
| |
| |
| def set_content_protection(self, enable): |
| """Sets the content protection state on the port. |
| |
| @param enable: True to enable; False to disable. |
| """ |
| self.chameleond_proxy.SetContentProtection(self.port_id, enable) |
| |
| |
| def is_content_protection_enabled(self): |
| """Returns True if the content protection is enabled on the port. |
| |
| @return: True if the content protection is enabled; otherwise, False. |
| """ |
| return self.chameleond_proxy.IsContentProtectionEnabled(self.port_id) |
| |
| |
| def is_video_input_encrypted(self): |
| """Returns True if the video input on the port is encrypted. |
| |
| @return: True if the video input is encrypted; otherwise, False. |
| """ |
| return self.chameleond_proxy.IsVideoInputEncrypted(self.port_id) |
| |
| |
| def start_monitoring_audio_video_capturing_delay(self): |
| """Starts an audio/video synchronization utility.""" |
| self.chameleond_proxy.StartMonitoringAudioVideoCapturingDelay() |
| |
| |
| def get_audio_video_capturing_delay(self): |
| """Gets the time interval between the first audio/video cpatured data. |
| |
| @return: A floating points indicating the time interval between the |
| first audio/video data captured. If the result is negative, |
| then the first video data is earlier, otherwise the first |
| audio data is earlier. |
| """ |
| return self.chameleond_proxy.GetAudioVideoCapturingDelay() |
| |
| |
| def start_capturing_video(self, box=None): |
| """ |
| Captures video frames. Asynchronous, returns immediately. |
| |
| @param box: int tuple, (x, y, width, height) pixel coordinates. |
| Defines the rectangular boundary within which to capture. |
| """ |
| |
| if box is None: |
| self.chameleond_proxy.StartCapturingVideo(self.port_id) |
| else: |
| self.chameleond_proxy.StartCapturingVideo(self.port_id, *box) |
| |
| |
| def stop_capturing_video(self): |
| """ |
| Stops the ongoing video frame capturing. |
| |
| """ |
| self.chameleond_proxy.StopCapturingVideo() |
| |
| |
| def get_captured_frame_count(self): |
| """ |
| @return: int, the number of frames that have been captured. |
| |
| """ |
| return self.chameleond_proxy.GetCapturedFrameCount() |
| |
| |
| def read_captured_frame(self, index): |
| """ |
| @param index: int, index of the desired captured frame. |
| @return: xmlrpclib.Binary object containing a byte-array of the pixels. |
| |
| """ |
| |
| frame = self.chameleond_proxy.ReadCapturedFrame(index) |
| return Image.fromstring('RGB', |
| self.get_captured_resolution(), |
| frame.data) |
| |
| |
| def get_captured_checksums(self, start_index=0, stop_index=None): |
| """ |
| @param start_index: int, index of the frame to start with. |
| @param stop_index: int, index of the frame (excluded) to stop at. |
| @return: a list of checksums of frames captured. |
| |
| """ |
| return self.chameleond_proxy.GetCapturedChecksums(start_index, |
| stop_index) |
| |
| |
| def get_captured_fps_list(self, time_to_start=0, total_period=None): |
| """ |
| @param time_to_start: time in second, support floating number, only |
| measure the period starting at this time. |
| If negative, it is the time before stop, e.g. |
| -2 meaning 2 seconds before stop. |
| @param total_period: time in second, integer, the total measuring |
| period. If not given, use the maximum time |
| (integer) to the end. |
| @return: a list of fps numbers, or [-1] if any error. |
| |
| """ |
| checksums = self.get_captured_checksums() |
| |
| frame_to_start = int(round(time_to_start * self._FRAME_RATE)) |
| if total_period is None: |
| # The default is the maximum time (integer) to the end. |
| total_period = (len(checksums) - frame_to_start) // self._FRAME_RATE |
| frame_to_stop = frame_to_start + total_period * self._FRAME_RATE |
| |
| if frame_to_start >= len(checksums) or frame_to_stop >= len(checksums): |
| logging.error('The given time interval is out-of-range.') |
| return [-1] |
| |
| # Only pick the checksum we are interested. |
| checksums = checksums[frame_to_start:frame_to_stop] |
| |
| # Count the unique checksums per second, i.e. FPS |
| logging.debug('Output the fps info below:') |
| fps_list = [] |
| for i in range(0, len(checksums), self._FRAME_RATE): |
| unique_count = 0 |
| debug_str = '' |
| for j in range(i, i + self._FRAME_RATE): |
| if j == 0 or checksums[j] != checksums[j - 1]: |
| unique_count += 1 |
| debug_str += '*' |
| else: |
| debug_str += '.' |
| fps_list.append(unique_count) |
| logging.debug('%2dfps %s', unique_count, debug_str) |
| |
| return fps_list |
| |
| |
| def search_fps_pattern(self, pattern_diff_frame, pattern_window=None, |
| time_to_start=0): |
| """Search the captured frames and return the time where FPS is greater |
| than given FPS pattern. |
| |
| A FPS pattern is described as how many different frames in a sliding |
| window. For example, 5 differnt frames in a window of 60 frames. |
| |
| @param pattern_diff_frame: number of different frames for the pattern. |
| @param pattern_window: number of frames for the sliding window. Default |
| is 1 second. |
| @param time_to_start: time in second, support floating number, |
| start to search from the given time. |
| @return: the time matching the pattern. -1.0 if not found. |
| |
| """ |
| if pattern_window is None: |
| pattern_window = self._FRAME_RATE |
| |
| checksums = self.get_captured_checksums() |
| |
| frame_to_start = int(round(time_to_start * self._FRAME_RATE)) |
| first_checksum = checksums[frame_to_start] |
| |
| for i in range(frame_to_start + 1, len(checksums) - pattern_window): |
| unique_count = 0 |
| for j in range(i, i + pattern_window): |
| if j == 0 or checksums[j] != checksums[j - 1]: |
| unique_count += 1 |
| if unique_count >= pattern_diff_frame: |
| return float(i) / self._FRAME_RATE |
| |
| return -1.0 |
| |
| |
| def get_captured_resolution(self): |
| """ |
| @return: (width, height) tuple, the resolution of captured frames. |
| |
| """ |
| return self.chameleond_proxy.GetCapturedResolution() |
| |
| |
| |
| class ChameleonAudioInput(ChameleonPort): |
| """ChameleonAudioInput is an abstraction of an audio input port. |
| |
| It contains some special methods to control an audio input. |
| """ |
| |
| def __init__(self, chameleon_port): |
| """Construct a ChameleonAudioInput. |
| |
| @param chameleon_port: A general ChameleonPort object. |
| """ |
| self.chameleond_proxy = chameleon_port.chameleond_proxy |
| self.port_id = chameleon_port.port_id |
| |
| |
| def start_capturing_audio(self): |
| """Starts capturing audio.""" |
| return self.chameleond_proxy.StartCapturingAudio(self.port_id) |
| |
| |
| def stop_capturing_audio(self): |
| """Stops capturing audio. |
| |
| Returns: |
| A tuple (remote_path, format). |
| remote_path: The captured file path on Chameleon. |
| format: A dict containing: |
| file_type: 'raw' or 'wav'. |
| sample_format: 'S32_LE' for 32-bit signed integer in little-endian. |
| Refer to aplay manpage for other formats. |
| channel: channel number. |
| rate: sampling rate. |
| """ |
| remote_path, data_format = self.chameleond_proxy.StopCapturingAudio( |
| self.port_id) |
| return remote_path, data_format |
| |
| |
| class ChameleonAudioOutput(ChameleonPort): |
| """ChameleonAudioOutput is an abstraction of an audio output port. |
| |
| It contains some special methods to control an audio output. |
| """ |
| |
| def __init__(self, chameleon_port): |
| """Construct a ChameleonAudioOutput. |
| |
| @param chameleon_port: A general ChameleonPort object. |
| """ |
| self.chameleond_proxy = chameleon_port.chameleond_proxy |
| self.port_id = chameleon_port.port_id |
| |
| |
| def start_playing_audio(self, path, data_format): |
| """Starts playing audio. |
| |
| @param path: The path to the file to play on Chameleon. |
| @param data_format: A dict containing data format. Currently Chameleon |
| only accepts data format: |
| dict(file_type='raw', sample_format='S32_LE', |
| channel=8, rate=48000). |
| |
| """ |
| self.chameleond_proxy.StartPlayingAudio(self.port_id, path, data_format) |
| |
| |
| def stop_playing_audio(self): |
| """Stops capturing audio.""" |
| self.chameleond_proxy.StopPlayingAudio(self.port_id) |
| |
| |
| def make_chameleon_hostname(dut_hostname): |
| """Given a DUT's hostname, returns the hostname of its Chameleon. |
| |
| @param dut_hostname: Hostname of a DUT. |
| |
| @return Hostname of the DUT's Chameleon. |
| """ |
| host_parts = dut_hostname.split('.') |
| host_parts[0] = host_parts[0] + '-chameleon' |
| return '.'.join(host_parts) |
| |
| |
| def make_btpeer_hostnames(dut_hostname): |
| """Given a DUT's hostname, returns the hostname of its bluetooth peers. |
| |
| A DUT can have up to 4 Bluetooth peers named hostname-btpeer[1-4] |
| @param dut_hostname: Hostname of a DUT. |
| |
| @return List of hostname of the DUT's Bluetooth peer devices |
| """ |
| hostnames = [] |
| host_parts = dut_hostname.split('.') |
| for i in range(1,5): |
| hostname_prefix = host_parts[0] + '-btpeer' +str(i) |
| hostname = [hostname_prefix] |
| hostname.extend(host_parts[1:]) |
| hostnames.append('.'.join(hostname)) |
| return hostnames |
| |
| def create_chameleon_board(dut_hostname, args): |
| """Creates a ChameleonBoard object with either DUT's hostname or arguments. |
| |
| If the DUT's hostname is in the lab zone, it connects to the Chameleon by |
| append the hostname with '-chameleon' suffix. If not, checks if the args |
| contains the key-value pair 'chameleon_host=IP'. |
| |
| @param dut_hostname: Hostname of a DUT. |
| @param args: A string of arguments passed from the command line. |
| |
| @return A ChameleonBoard object. |
| |
| @raise ChameleonConnectionError if unknown hostname. |
| """ |
| connection = None |
| hostname = make_chameleon_hostname(dut_hostname) |
| if utils.host_is_in_lab_zone(hostname): |
| connection = ChameleonConnection(hostname) |
| else: |
| args_dict = utils.args_to_dict(args) |
| hostname = args_dict.get('chameleon_host', None) |
| port = args_dict.get('chameleon_port', CHAMELEON_PORT) |
| if hostname: |
| connection = ChameleonConnection(hostname, port) |
| else: |
| raise ChameleonConnectionError('No chameleon_host is given in args') |
| |
| return ChameleonBoard(connection) |
