chameleond/drivers/fpga_tio.py

# -*- coding: utf-8 -*-
# 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.
"""Chameleond Driver for FPGA customized platform with the TIO card."""

from __future__ import print_function

import functools
import glob
# TODO: to port chromite.lib.cros_logging to replace legacy logging
import logging  # pylint: disable=cros-logging-import
import os
import time
import xmlrpc.client

from . import chameleon_common  # pylint: disable=W0611

from chameleond.devices import audio_board
from chameleond.devices import avsync_probe
from chameleond.devices import bluetooth_base_flow
from chameleond.devices import bluetooth_hid_flow
from chameleond.devices import codec_flow
from chameleond.devices import input_flow
from chameleond.devices import motor_board
from chameleond.devices import usb_audio_flow
from chameleond.devices import usb_hid_flow
from chameleond.devices import usb_printer_device
from chameleond.interface import ChameleondInterface
from chameleond.utils import caching_server
from chameleond.utils import device_manager
from chameleond.utils import fpga
from chameleond.utils import flow_manager
from chameleond.utils import i2c
from chameleond.utils import ids
from chameleond.utils import system_tools
from chameleond.utils import usb
from chameleond.utils import usb_printer_control
from chameleond.utils.common import lazy
from chameleond.utils import bluetooth_a2dp

if os.environ.get('PLATFORM') != 'FPGA':
  # pylint: disable=C0412
  from chameleond.utils import bluetooth_tester

if os.environ.get('PLATFORM') == 'RASPI':
  # pylint: disable=C0412
  from chameleond.devices import raspi_bluetooth_flow


class DriverError(Exception):
  """Exception raised when any error on FPGA driver."""
  pass


def _DeviceMethod(device_id):
  """Decorator that checks if the device exists.

  Args:
    device_id: The device ID.
  """
  def _ActualDecorator(func):
    @functools.wraps(func)
    def wrapper(instance, *args, **kwargs):
      if not instance.HasDevice(device_id):
        raise DriverError('There is no %s' % ids.DEVICE_NAMES[device_id])
      return func(instance, *args, **kwargs)
    return wrapper
  return _ActualDecorator


class ChameleondDriver(ChameleondInterface):
  """Chameleond Driver for FPGA customized platform."""

  _I2C_BUS_MAIN = 0
  _I2C_BUS_AUDIO_CODEC = 1
  _I2C_BUS_EXT_BOARD = 3

  # Time to wait for video frame dump to start before a timeout error is raised
  _TIMEOUT_FRAME_DUMP_PROBE = 60.0

  # The frame index which is used for the regular DumpPixels API.
  _DEFAULT_FRAME_INDEX = 0
  _DEFAULT_FRAME_LIMIT = _DEFAULT_FRAME_INDEX + 1

  # Limit the period of async capture to 3min (in 60fps).
  _MAX_CAPTURED_FRAME_COUNT = 3 * 60 * 60

  def __init__(self, *args, **kwargs):
    super(ChameleondDriver, self).__init__(*args, **kwargs)

    self._platform = os.environ.get('PLATFORM')
    logging.info('platform: %s', self._platform)

    self.dump_bt_peer_info()

    self._captured_params = {}
    self._process = None

    # waihong@chromium.org suggests to use a lazy wrapper which instantiates
    # the following control objects when requested at the first time.
    self._main_bus = lazy(i2c.I2cBus)(self._I2C_BUS_MAIN)
    self._ext_board_bus = lazy(i2c.I2cBus)(self._I2C_BUS_EXT_BOARD)
    self._audio_codec_bus = lazy(i2c.I2cBus)(self._I2C_BUS_AUDIO_CODEC)
    self._fpga_ctrl = lazy(fpga.FpgaController)()
    self._usb_audio_ctrl = lazy(usb.USBAudioController)()
    self._usb_hid_ctrl = lazy(usb.USBController)('g_hid')
    self._usb_printer_ctrl = lazy(usb_printer_control.USBPrinterController)()
    self._bluetooth_hid_ctrl = lazy(usb.USBController)(
        bluetooth_hid_flow.BluetoothHIDMouseFlow.DRIVER)
    self._bluetooth_a2dp_sink_ctrl = lazy(usb.USBController)(
        bluetooth_a2dp.BluetoothA2DPSinkFlow.DRIVER)
    # See explanation for using DRIVER_MODULE in bluetooth_nrf52.py
    self._ble_hid_ctrl = lazy(usb.USBController)(
        bluetooth_hid_flow.BleHIDMouseFlow.DRIVER_MODULE)
    # A base control comes without any driver.
    self._base_ctrl = lazy(usb.USBController)('')

    if self._platform == 'CHROMEOS':
      self._devices = self.init_devices_for_chromeos()
    elif self._platform == 'RASPI':
      self._devices = self.init_devices_for_raspi()
    else:
      self._devices = self.init_devices_for_fpga()

    self._device_manager = device_manager.DeviceManager(self._devices)
    self._device_manager.Init()
    self._flows = self._device_manager.GetDetectedFlows()

    # Allow to access the methods through object.
    # Hence, there is no need to export the methods in ChameleondDriver.
    # An object in the following would be None if it is not instantiated
    # in self._devices above.
    self.audio_board = self._device_manager.GetChameleonDevice(ids.AUDIO_BOARD)
    self.bluetooth_audio = self._device_manager.GetChameleonDevice(
        ids.BLUETOOTH_AUDIO)
    self.bluetooth_base = self._device_manager.GetChameleonDevice(
        ids.BLUETOOTH_BASE)
    self.bluetooth_mouse = self._device_manager.GetChameleonDevice(
        ids.BLUETOOTH_HID_MOUSE)
    self.bluetooth_keyboard = self._device_manager.GetChameleonDevice(
        ids.BLUETOOTH_HID_KEYBOARD)
    self.bluetooth_tester = self._device_manager.GetChameleonDevice(
        ids.BLUETOOTH_TESTER)
    self.avsync_probe = self._device_manager.GetChameleonDevice(
        ids.AVSYNC_PROBE)
    self.motor_board = self._device_manager.GetChameleonDevice(ids.MOTOR_BOARD)
    self.printer = self._device_manager.GetChameleonDevice(ids.USB_PRINTER)
    self.bluetooth_a2dp_sink = self._device_manager.GetChameleonDevice(
        ids.BLUETOOTH_A2DP_SINK)
    self.ble_mouse = self._device_manager.GetChameleonDevice(
        ids.BLE_MOUSE)
    self.ble_keyboard = self._device_manager.GetChameleonDevice(
        ids.BLE_KEYBOARD)
    self.ble_phone = self._device_manager.GetChameleonDevice(
        ids.BLE_PHONE)
    self._flow_manager = flow_manager.FlowManager(self._flows)

    self.Reset()


  def get_platform(self):
    return self._platform

  def init_devices_for_chromeos(self):
    devices = {
        ids.BLUETOOTH_HID_MOUSE:
            bluetooth_hid_flow.BluetoothHIDMouseFlow(
                ids.BLUETOOTH_HID_MOUSE, self._bluetooth_hid_ctrl),
        ids.BLUETOOTH_A2DP_SINK:
            bluetooth_a2dp.BluetoothA2DPSinkFlow(
                ids.BLUETOOTH_A2DP_SINK, self._bluetooth_a2dp_sink_ctrl),
        ids.BLE_MOUSE:
            bluetooth_hid_flow.BleHIDMouseFlow(
                ids.BLE_MOUSE, self._ble_hid_ctrl),
        ids.BLUETOOTH_BASE:
            bluetooth_base_flow.BluetoothBaseFlow(
                ids.BLUETOOTH_BASE, self._base_ctrl),
    }
    return devices

  def init_devices_for_fpga(self):
    devices = {
        ids.DP1: input_flow.DpInputFlow(
            ids.DP1, self._main_bus, self._fpga_ctrl),
        ids.DP2: input_flow.DpInputFlow(
            ids.DP2, self._main_bus, self._fpga_ctrl),
        ids.HDMI: input_flow.HdmiInputFlow(
            ids.HDMI, self._main_bus, self._fpga_ctrl),
        ids.VGA: input_flow.VgaInputFlow(
            ids.VGA, self._main_bus, self._fpga_ctrl),
        ids.MIC: codec_flow.InputCodecFlow(
            ids.MIC, self._audio_codec_bus, self._fpga_ctrl),
        ids.LINEIN: codec_flow.InputCodecFlow(
            ids.LINEIN, self._audio_codec_bus, self._fpga_ctrl),
        ids.LINEOUT: codec_flow.OutputCodecFlow(
            ids.LINEOUT, self._audio_codec_bus, self._fpga_ctrl),
        ids.USB_AUDIO_IN: usb_audio_flow.InputUSBAudioFlow(
            ids.USB_AUDIO_IN, self._usb_audio_ctrl),
        ids.USB_AUDIO_OUT: usb_audio_flow.OutputUSBAudioFlow(
            ids.USB_AUDIO_OUT, self._usb_audio_ctrl),
        ids.USB_KEYBOARD: usb_hid_flow.KeyboardUSBHIDFlow(
            ids.USB_KEYBOARD, self._usb_hid_ctrl),
        ids.USB_TOUCH: usb_hid_flow.TouchUSBHIDFlow(
            ids.USB_TOUCH, self._usb_hid_ctrl),
        ids.AVSYNC_PROBE: avsync_probe.AVSyncProbe(ids.AVSYNC_PROBE),
        ids.AUDIO_BOARD: audio_board.AudioBoard(self._ext_board_bus),
        ids.MOTOR_BOARD: motor_board.MotorBoard(self._ext_board_bus),
        ids.USB_PRINTER: usb_printer_device.USBPrinter(self._usb_printer_ctrl),
    }

    return devices

  def init_devices_for_raspi(self):
    devices = {
        ids.BLUETOOTH_HID_KEYBOARD: raspi_bluetooth_flow.RaspiHIDKeyboard(),
        ids.BLUETOOTH_HID_MOUSE: raspi_bluetooth_flow.RaspiHIDMouse(),
        ids.BLE_KEYBOARD: raspi_bluetooth_flow.RaspiHIDKeyboard(),
        ids.BLE_MOUSE: raspi_bluetooth_flow.RaspiHIDMouse(),
        ids.BLUETOOTH_BASE:
            bluetooth_base_flow.BluetoothBaseFlow(
                ids.BLUETOOTH_BASE, self._base_ctrl),
        ids.BLUETOOTH_TESTER: bluetooth_tester.BluetoothTester(),
        ids.BLE_PHONE: raspi_bluetooth_flow.RaspiBLEPhone(),
        ids.BLUETOOTH_AUDIO: raspi_bluetooth_flow.RaspiBluetoothAudioFlow(),
    }
    return devices

  def Reset(self):
    """Resets Chameleon board."""
    logging.info('Execute the reset process')
    self._flow_manager.Reset()
    self._device_manager.Reset()

    self._ClearAudioFiles()
    self._ClearPrinterFiles()
    caching_server.ClearCachedDir()

  def Reboot(self):
    """Reboots Chameleon board."""
    logging.info('The chameleon board is going to reboot.')
    system_tools.SystemTools.Call('reboot')

  def dump_bt_peer_info(self):
    """Dump the kernel, BlueZ, chameleon bundle info"""
    kernel_info = system_tools.SystemTools.Output('uname', '-a').strip()
    bluez_info = system_tools.SystemTools.Output('bluetoothd', '-v').strip()

    logging.info('kernel_info: %s', kernel_info)
    logging.info('bluez_version: %s', bluez_info)
    logging.info('bt_pkg_commit: %s', self.get_bt_commit_hash())

  def get_bt_commit_hash(self):
    """Return the last commit hash in chameleon_commits."""
    try:
      output = system_tools.SystemTools.Output('grep', 'BUNDLE_COMMIT_HASH',
          '/etc/default/chameleond')
      logging.debug('output is %s', output)
      commit = output.split('=')[1].strip().replace('"', '')
      logging.debug('got commit %s', commit)
      return commit
    except Exception as e:
      logging.error('Exception %s in get_bt_commit_hash', str(e))
      return ''

  def log_message(self, msg):
    """Write the msg to the chamelond log and system log."""
    try:
      logging.info(msg)
      system_tools.SystemTools.Call('logger', msg)
    except Exception as e:
      logging.error("Exception '%s' in log_message msg '%s'", str(e), msg)

  def GetDetectedStatus(self):
    """Returns detetcted status of all devices.

    User can use this API to know the capability of the chameleon board.

    Returns:
      A list of a tuple of detected devices' strings detected status.
      e.g. [('HDMI', True), ('MIC', False)]
    """
    detected_list = []
    for device_id in self._devices:
      detected = False
      if self._device_manager.GetChameleonDevice(device_id):
        detected = True
      detected_list.append((ids.DEVICE_NAMES[device_id], detected))
    return detected_list

  def GetSupportedPorts(self):
    """Returns all supported ports on the board.

    Not like the ProbePorts() method which only returns the ports which
    are connected, this method returns all supported ports on the board.

    Returns:
      A tuple of port_id, for all supported ports on the board.
    """
    return self._flow_manager.GetSupportedPorts()

  def GetSupportedInputs(self):
    """Returns all supported input ports on the board.

    Not like the ProbeInputs() method which only returns the input ports which
    are connected, this method returns all supported input ports on the board.

    Returns:
      A tuple of port_id, for all supported input port on the board.
    """
    return self._flow_manager.GetSupportedInputs()

  def GetSupportedOutputs(self):
    """Returns all supported output ports on the board.

    Not like the ProbeOutputs() method which only returns the output ports which
    are connected, this method returns all supported output ports on the board.

    Returns:
      A tuple of port_id, for all supported output port on the board.
    """
    return self._flow_manager.GetSupportedOutputs()

  def IsPhysicalPlugged(self, port_id):
    """Returns true if the physical cable is plugged between DUT and Chameleon.

    Args:
      port_id: The ID of the input/output port.

    Returns:
      True if the physical cable is plugged; otherwise, False.
    """
    return self._flow_manager.IsPhysicalPlugged(port_id)

  def ProbePorts(self):
    """Probes all the connected ports on Chameleon board.

    Returns:
      A tuple of port_id, for the ports connected to DUT.
    """
    return self._flow_manager.ProbePorts()

  def ProbeInputs(self):
    """Probes all the connected input ports on Chameleon board.

    Returns:
      A tuple of port_id, for the input ports connected to DUT.
    """
    return self._flow_manager.ProbeInputs()

  def ProbeOutputs(self):
    """Probes all the connected output ports on Chameleon board.

    Returns:
      A tuple of port_id, for the output ports connected to DUT.
    """
    return self._flow_manager.ProbeOutputs()

  def GetConnectorType(self, port_id):
    """Returns the human readable string for the connector type.

    Args:
      port_id: The ID of the input/output port.

    Returns:
      A string, like "HDMI", "DP", "MIC", etc.
    """
    return self._flow_manager.GetConnectorType(port_id)

  def HasAudioSupport(self, port_id):
    """Returns true if the port has audio support.

    Args:
      port_id: The ID of the input/output port.

    Returns:
      True if the input/output port has audio support; otherwise, False.
    """
    return self._flow_manager.HasAudioSupport(port_id)

  def HasVideoSupport(self, port_id):
    """Returns true if the port has video support.

    Args:
      port_id: The ID of the input/output port.

    Returns:
      True if the input/output port has video support; otherwise, False.
    """
    return self._flow_manager.HasVideoSupport(port_id)

  def SetVgaMode(self, port_id, mode):
    """Sets the mode for VGA monitor.

    Args:
      port_id: The ID of the VGA port.
      mode: A string of the mode name, e.g. 'PC_1920x1080x60'. Use 'auto'
            to detect the VGA mode automatically.
    """
    return self._flow_manager.SetVgaMode(port_id, mode)

  def WaitVideoInputStable(self, port_id, timeout=None):
    """Waits the video input stable or timeout.

    Args:
      port_id: The ID of the video input port.
      timeout: The time period to wait for.

    Returns:
      True if the video input becomes stable within the timeout period;
      otherwise, False.
    """
    return self._flow_manager.WaitVideoInputStable(port_id, timeout)

  def CreateEdid(self, edid):
    """Creates an internal record of EDID using the given byte array.

    Args:
      edid: A byte array of EDID data, wrapped in a xmlrpclib.Binary object.

    Returns:
      An edid_id.
    """
    return self._flow_manager.CreateEdid(edid)

  def DestroyEdid(self, edid_id):
    """Destroys the internal record of EDID. The internal data will be freed.

    Args:
      edid_id: The ID of the EDID, which was created by CreateEdid().
    """
    self._flow_manager.DestroyEdid(edid_id)

  def SetDdcState(self, port_id, enabled):
    """Sets the enabled/disabled state of DDC bus on the given video input.

    Args:
      port_id: The ID of the video input port.
      enabled: True to enable DDC bus due to an user request; False to
               disable it.
    """
    self._flow_manager.SetDdcState(port_id, enabled)

  def IsDdcEnabled(self, port_id):
    """Checks if the DDC bus is enabled or disabled on the given video input.

    Args:
      port_id: The ID of the video input port.

    Returns:
      True if the DDC bus is enabled; False if disabled.
    """
    return self._flow_manager.IsDdcEnabled(port_id)

  def ReadEdid(self, port_id):
    """Reads the EDID content of the selected video input on Chameleon.

    Args:
      port_id: The ID of the video input port.

    Returns:
      A byte array of EDID data, wrapped in a xmlrpclib.Binary object,
      or None if the EDID is disabled.
    """
    return self._flow_manager.ReadEdid(port_id)

  def ApplyEdid(self, port_id, edid_id):
    """Applies the EDID to the selected video input.

    Note that this method doesn't pulse the HPD line. Should call Plug(),
    Unplug(), or FireHpdPulse() later.

    Args:
      port_id: The ID of the video input port.
      edid_id: The ID of the EDID.
    """
    self._flow_manager.ApplyEdid(port_id, edid_id)

  def IsPlugged(self, port_id):
    """Returns true if the port is emulated as plugged.

    Args:
      port_id: The ID of the input/output port.

    Returns:
      True if the port is emualted as plugged; otherwise, False.
    """
    return self._flow_manager.IsPlugged(port_id)

  def Plug(self, port_id):
    """Emualtes plug, like asserting HPD line to high on a video port.

    Args:
      port_id: The ID of the input/output port.
    """
    return self._flow_manager.Plug(port_id)

  def Unplug(self, port_id):
    """Emulates unplug, like deasserting HPD line to low on a video port.

    Args:
      port_id: The ID of the input/output port.
    """
    return self._flow_manager.Unplug(port_id)

  def UnplugHPD(self, port_id):
    """Only deassert HPD line to low on a video port.

    Args:
      port_id: The ID of the input/output port.
    """
    return self._flow_manager.UnplugHPD(port_id)

  def FireHpdPulse(self, port_id, deassert_interval_usec,
                   assert_interval_usec=None, repeat_count=1,
                   end_level=1):
    """Fires one or more HPD pulse (low -> high -> low -> ...).

    Args:
      port_id: The ID of the video input port.
      deassert_interval_usec: The time in microsecond of the deassert pulse.
      assert_interval_usec: The time in microsecond of the assert pulse.
                            If None, then use the same value as
                            deassert_interval_usec.
      repeat_count: The count of HPD pulses to fire.
      end_level: HPD ends with 0 for LOW (unplugged) or 1 for HIGH (plugged).
    """
    return self._flow_manager.FireHpdPulse(
        port_id, deassert_interval_usec, assert_interval_usec, repeat_count,
        end_level)

  def FireMixedHpdPulses(self, port_id, widths_msec):
    """Fires one or more HPD pulses, starting at low, of mixed widths.

    One must specify a list of segment widths in the widths_msec argument where
    widths_msec[0] is the width of the first low segment, widths_msec[1] is that
    of the first high segment, widths_msec[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.

    The method is equivalent to a series of calls to Unplug() and Plug()
    separated by specified pulse widths.

    Args:
      port_id: The ID of the video input port.
      widths_msec: list of pulse segment widths in milli-second.
    """
    return self._flow_manager.FireMixedHpdPulses(port_id, widths_msec)

  def ScheduleHpdToggle(self, port_id, delay_ms, rising_edge):
    """Schedules one HPD Toggle, with a delay between the toggle.

    Args:
      port_id: The ID of the video input port.
      delay_ms: Delay in milli-second before the toggle takes place.
      rising_edge: Whether the toggle should be a rising edge or a falling edge.
    """
    return self._flow_manager.ScheduleHpdToggle(port_id, delay_ms, rising_edge)

  def SetContentProtection(self, port_id, enabled):
    """Sets the content protection state on the port.

    Args:
      port_id: The ID of the video input port.
      enabled: True to enable; False to disable.
    """
    self._flow_manager.SetContentProtection(port_id, enabled)

  def IsContentProtectionEnabled(self, port_id):
    """Returns True if the content protection is enabled on the port.

    Args:
      port_id: The ID of the video input port.

    Returns:
      True if the content protection is enabled; otherwise, False.
    """
    return self._flow_manager.IsContentProtectionEnabled(port_id)

  def IsVideoInputEncrypted(self, port_id):
    """Returns True if the video input on the port is encrypted.

    Args:
      port_id: The ID of the video input port.

    Returns:
      True if the video input is encrypted; otherwise, False.
    """
    return self._flow_manager.IsVideoInputEncrypted(port_id)

  def StartMonitoringAudioVideoCapturingDelay(self):
    """Starts an audio/video synchronization utility

    The example of usage:
      chameleon.StartMonitoringAudioVideoCapturingDelay()
      chameleon.StartCapturingVideo(hdmi_input)
      chameleon.StartCapturingAudio(hdmi_input)
      time.sleep(2)
      chameleon.StopCapturingVideo()
      chameleon.StopCapturingAudio(hdmi_input)
      delay = chameleon.GetAudioVideoCapturingDelay()
    """
    self._process = system_tools.SystemTools.RunInSubprocess('avsync')

  def GetAudioVideoCapturingDelay(self):
    """Get the time interval between the first audio/video cpatured data

    Returns:
      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.

    Raises:
      DriverError if there is no output from the monitoring process.
    """

    if self._process.poll() is None:
      self._process.terminate()
      raise DriverError('The monitoring process has not finished.')

    return_code, out, err = system_tools.SystemTools.GetSubprocessOutput(
        self._process)

    if return_code != 0 or err:
      raise DriverError('Runtime error in the monitoring process')

    if not out:
      raise DriverError('No output from the monitoring process.')

    return float(out)

  def DumpPixels(self, port_id, x=None, y=None, width=None, height=None):
    """Dumps the raw pixel array of the selected area.

    If not given the area, default to capture the whole screen.

    Args:
      port_id: The ID of the video input port.
      x: The X position of the top-left corner.
      y: The Y position of the top-left corner.
      width: The width of the area.
      height: The height of the area.

    Returns:
      A byte-array of the pixels, wrapped in a xmlrpclib.Binary object.
    """
    x, y, width, height = self._AutoFillArea(port_id, x, y, width, height)
    self.CaptureVideo(port_id, self._DEFAULT_FRAME_LIMIT, x, y, width, height)
    return self.ReadCapturedFrame(self._DEFAULT_FRAME_INDEX)

  def _AutoFillArea(self, port_id, x, y, width, height):
    """Verifies the area argument correctness and fills the default values.

    It keeps x=None and y=None if all of the x, y, width, and height are None.
    That hints FPGA to use a full-screen capture, not a cropped-sccren capture.

    Args:
      port_id: The ID of the video input port.
      x: The X position of the top-left corner.
      y: The Y position of the top-left corner.
      width: The width of the area.
      height: The height of the area.

    Returns:
      A tuple of (x, y, width, height)

    Raises:
      DriverError if the area is not specified correctly.
    """
    if (x, y, width, height) == (None, ) * 4:
      return (None, None) + self.DetectResolution(port_id)
    elif (x, y) == (None, ) * 2 or None not in (x, y, width, height):
      return (x, y, width, height)
    else:
      raise DriverError('Some of area arguments are not specified.')

  def GetMaxFrameLimit(self, port_id, width, height):
    """Gets the maximal number of frames which are accommodated in the buffer.

    It depends on the size of the internal buffer on the board and the
    size of area to capture (full screen or cropped area).

    Args:
      port_id: The ID of the video input port.
      width: The width of the area to capture.
      height: The height of the area to capture.

    Returns:
      A number of the frame limit.
    """
    return self._flow_manager.GetMaxFrameLimit(port_id, width, height)

  def _PrepareCapturingVideo(self, port_id, width, height):
    """Prepares capturing video on the given video input.

    Args:
      port_id: The ID of the video input port.
      width: The width of the area of crop.
      height: The height of the area of crop.
    """
    caching_server.ClearCachedDir()
    self._flow_manager.SelectInput(port_id)
    if not self._flow_manager.IsPlugged(port_id):
      raise DriverError('HPD is unplugged. No signal is expected.')
    self._captured_params = {
        'port_id': port_id,
        'max_frame_limit': self._flow_manager.GetMaxFrameLimit(port_id,
                                                               width, height)
    }

  def StartCapturingVideo(self, port_id, x=None, y=None, width=None,
                          height=None):
    """Starts video capturing continuously on the given video input.

    This API is an asynchronous call. It returns after the video starts
    capturing. The caller should call StopCapturingVideo to stop it.

    The example of usage:
      chameleon.StartCapturingVideo(hdmi_input)
      time.sleep(2)
      chameleon.StopCapturingVideo()
      for i in xrange(chameleon.GetCapturedFrameCount()):
        frame = chameleon.ReadCapturedFrame(i, *area).data
        CompareFrame(frame, golden_frames[i])

    Args:
      port_id: The ID of the video input port.
      x: The X position of the top-left corner of crop.
      y: The Y position of the top-left corner of crop.
      width: The width of the area of crop.
      height: The height of the area of crop.
    """
    x, y, width, height = self._AutoFillArea(port_id, x, y, width, height)
    self._PrepareCapturingVideo(port_id, width, height)
    max_frame_limit = self._captured_params['max_frame_limit']
    logging.info('Start capturing video from port #%d', port_id)
    self._flow_manager.StartDumpingFrames(
        port_id, max_frame_limit, x, y, width, height,
        self._MAX_CAPTURED_FRAME_COUNT)

  def StopCapturingVideo(self, stop_index=None):
    """Stops video capturing which was started previously.

    Args:
      stop_index: Wait for the captured frame count to reach this index. If
                  not given, stop immediately. Note that the captured frame of
                  stop_index should not be read.

    Raises:
      DriverError if the capture period is longer than the capture limitation.
    """
    port_id = self._captured_params['port_id']
    if stop_index:
      if stop_index >= self._MAX_CAPTURED_FRAME_COUNT:
        raise DriverError('Exceeded the limit of capture, stop_index >= %d' %
                          self._MAX_CAPTURED_FRAME_COUNT)
      logging.info('Waiting the captured frame count reaches %d...', stop_index)
      while self.GetCapturedFrameCount() < stop_index:
        pass

    self._flow_manager.StopDumpingFrames(port_id)
    logging.info('Stopped capturing video from port #%d', port_id)
    if self.GetCapturedFrameCount() >= self._MAX_CAPTURED_FRAME_COUNT:
      raise DriverError('Exceeded the limit of capture, frame_count >= %d' %
                        self._MAX_CAPTURED_FRAME_COUNT)

  def CaptureVideo(self, port_id, total_frame, x=None, y=None, width=None,
                   height=None):
    """Captures the video stream on the given video input to the buffer.

    This API is a synchronous call. It returns after all the frames are
    captured. The frames can be read using the ReadCapturedFrame API.

    The example of usage:
      chameleon.CaptureVideo(hdmi_input, total_frame)
      for i in xrange(total_frame):
        frame = chameleon.ReadCapturedFrame(i, *area).data
        CompareFrame(frame, golden_frames[i])

    Args:
      port_id: The ID of the video input port.
      total_frame: The total number of frames to capture, should not larger
                   than value of GetMaxFrameLimit.
      x: The X position of the top-left corner of crop.
      y: The Y position of the top-left corner of crop.
      width: The width of the area of crop.
      height: The height of the area of crop.
    """
    x, y, width, height = self._AutoFillArea(port_id, x, y, width, height)
    logging.info('Capture video from port #%d', port_id)
    self._PrepareCapturingVideo(port_id, width, height)
    max_frame_limit = self._captured_params['max_frame_limit']
    if total_frame > max_frame_limit:
      raise DriverError('Exceed the max frame limit %d > %d' %
                        (total_frame, max_frame_limit))

    # TODO(waihong): Make the timeout value based on the FPS rate.
    self._flow_manager.DumpFramesToLimit(
        port_id, total_frame, x, y, width, height,
        self._TIMEOUT_FRAME_DUMP_PROBE)

  def GetCapturedFrameCount(self):
    """Gets the total count of the captured frames.

    Returns:
      The number of frames captured.
    """
    port_id = self._captured_params['port_id']
    return self._flow_manager.GetDumpedFrameCount(port_id)

  def GetCapturedResolution(self):
    """Gets the resolution of the captured frame.

    If a cropping area is specified on capturing, returns the cropped
    resolution.

    Returns:
      A (width, height) tuple.
    """
    port_id = self._captured_params['port_id']
    return self._flow_manager.GetCapturedResolution(port_id)

  def ReadCapturedFrame(self, frame_index):
    """Reads the content of the captured frame from the buffer.

    Args:
      frame_index: The index of the frame to read.

    Returns:
      A byte-array of the pixels, wrapped in a xmlrpclib.Binary object.
    """
    port_id = self._captured_params['port_id']
    total_frame = self.GetCapturedFrameCount()
    max_frame_limit = self._captured_params['max_frame_limit']
    # The captured frames are store in a circular buffer. Only the latest
    # max_frame_limit frames are valid.
    first_valid_index = max(0, total_frame - max_frame_limit)
    if not first_valid_index <= frame_index < total_frame:
      raise DriverError('The frame index is out-of-range: %d not in [%d, %d)' %
                        (frame_index, first_valid_index, total_frame))

    # Use the projected index.
    frame_index = frame_index % max_frame_limit
    screen = self._flow_manager.ReadCapturedFrame(port_id, frame_index)
    return xmlrpc.client.Binary(screen)

  def CacheFrameThumbnail(self, frame_index, ratio=2):
    """Caches the thumbnail of the dumped field to a temp file.

    Args:
      frame_index: The index of the frame to cache.
      ratio: The ratio to scale down the image.

    Returns:
      An ID to identify the cached thumbnail.
    """
    port_id = self._captured_params['port_id']
    return self._flow_manager.CacheFrameThumbnail(port_id, frame_index, ratio)

  def _GetCapturedSignals(self, signal_func_name, start_index=0,
                          stop_index=None):
    """Gets the list of signals of the captured frames.

    Args:
      signal_func_name: The name of the signal function, e.g. 'GetFrameHashes'.
      start_index: The index of the start frame. Default is 0.
      stop_index: The index of the stop frame (excluded). Default is the
                  value of GetCapturedFrameCount.

    Returns:
      The list of signals.
    """
    port_id = self._captured_params['port_id']
    total_frame = self.GetCapturedFrameCount()
    if stop_index is None:
      stop_index = total_frame
    if not 0 <= start_index < total_frame:
      raise DriverError('The start index is out-of-range: %d not in [0, %d)' %
                        (start_index, total_frame))
    if not 0 < stop_index <= total_frame:
      raise DriverError('The stop index is out-of-range: %d not in (0, %d]' %
                        (stop_index, total_frame))
    signal_func = getattr(self._flows[port_id], signal_func_name)
    return signal_func(start_index, stop_index)

  def GetCapturedChecksums(self, start_index=0, stop_index=None):
    """Gets the list of checksums of the captured frames.

    Args:
      start_index: The index of the start frame. Default is 0.
      stop_index: The index of the stop frame (excluded). Default is the
                  value of GetCapturedFrameCount.

    Returns:
      The list of checksums of frames.
    """
    return self._GetCapturedSignals('GetFrameHashes', start_index, stop_index)

  def GetCapturedHistograms(self, start_index=0, stop_index=None):
    """Gets the list of histograms of the captured frames.

    Args:
      start_index: The index of the start frame. Default is 0.
      stop_index: The index of the stop frame (excluded). Default is the
                  value of GetCapturedFrameCount.

    Returns:
      The list of histograms of frames.
    """
    return self._GetCapturedSignals('GetHistograms', start_index, stop_index)

  def ComputePixelChecksum(
      self, port_id, x=None, y=None, width=None, height=None):
    """Computes the checksum of pixels in the selected area.

    If not given the area, default to compute the whole screen.

    Args:
      port_id: The ID of the video input port.
      x: The X position of the top-left corner.
      y: The Y position of the top-left corner.
      width: The width of the area.
      height: The height of the area.

    Returns:
      The checksum of the pixels.
    """
    x, y, width, height = self._AutoFillArea(port_id, x, y, width, height)
    self.CaptureVideo(port_id, self._DEFAULT_FRAME_LIMIT, x, y, width, height)
    return self.GetCapturedChecksums(self._DEFAULT_FRAME_INDEX,
                                     self._DEFAULT_FRAME_INDEX + 1)[0]

  def DetectResolution(self, port_id):
    """Detects the video source resolution.

    Args:
      port_id: The ID of the video input port.

    Returns:
      A (width, height) tuple.
    """
    return self._flow_manager.DetectResolution(port_id)

  def GetVideoParams(self, port_id):
    """Gets video parameters.

    Args:
      port_id: The ID of the video input port.

    Returns:
      A dict containing video parameters. Fields are omitted if unknown.
    """
    return self._flow_manager.GetVideoParams(port_id)

  def GetLastInfoFrame(self, port_id, infoframe_type):
    """Obtains the last received InfoFrame of the specified type.

    Args:
      port_id: The ID of the video input port
      infoframe_type (string): the InfoFrame type

    Returns:
      A dict containing the InfoFrame.
    """
    infoframe = self._flow_manager.GetLastInfoFrame(port_id, infoframe_type)
    # Wrap the binary blob in an xmlrpclib object so that it can be serialized
    # in the XML-RPC reply.
    infoframe['payload'] = xmlrpc.client.Binary(infoframe['payload'])
    return infoframe

  def HasAudioBoard(self):
    """Returns True if there is an audio board.

    Returns:
      True if there is an audio board. False otherwise.
    """
    return self.audio_board is not None

  def HasDevice(self, device_id):
    """Returns True if there is a device.

    Returns:
      True if there is a device . False otherwise.
    """
    return self._device_manager.GetChameleonDevice(device_id) is not None

  def GetAudioChannelMapping(self, port_id):
    """Obtains the channel mapping for an audio port.

    Args:
      port_id: The ID of the audio port.

    Returns:
      An array of integers. There is one element per Chameleon channel.
      For audio input ports, each element indicates which input channel the
      capture channel is mapped to. For audio output ports, each element
      indicates which output channel the playback channel is mapped to. As a
      special case, -1 means the channel isn't mapped.

    Raises:
      FlowManagerError: no audio capture in progress
    """
    return self._flow_manager.GetAudioChannelMapping(port_id)

  def GetAudioFormat(self, port_id):
    """Gets the format currently used by audio capture.

    Args:
      port_id: The ID of the audio input port.

    Returns:
      A dict containing the format properties. The keys are:
      file_type: 'raw' or 'wav'
      sample_format: 'S32_LE' for 32-bit signed integers in little-endian. See
        aplay(1) for more formats.
      channel: number of channels
      rate: sampling rate in Hz (or zero if unknown)

    Raises:
      FlowManagerError: no audio capture in progress
    """
    return self._flow_manager.GetAudioFormat(port_id).AsDict()

  def TriggerLinkFailure(self, port_id):
    """Trigger a link failure on the port.

    Args:
      port_id: The ID of the input port.
    """
    self._flow_manager.TriggerLinkFailure(port_id)

  def StartCapturingAudio(self, port_id, has_file=True):
    """Starts capturing audio.

    Refer to the docstring of StartPlayingEcho about the restriction of
    capturing and echoing at the same time.

    Args:
      port_id: The ID of the audio input port.
      has_file: True for saving audio data to file. False otherwise.
    """
    self._flow_manager.StartCapturingAudio(port_id, has_file)

  def StopCapturingAudio(self, port_id):
    """Stops capturing audio and returns recorded data path and format.

    Args:
      port_id: The ID of the audio input port.

    Returns:
      A tuple (path, format).
      path: The path to the captured audio data.
      format: The format of the captured data. See GetAudioFormat. Note that
        the returned audio frequency may not be correct, for this reason
        calling GetAudioFormat during the capture is preferred.
      If we assign parameter has_file=False in StartCapturingAudio, we will get
      both empty string in path and format.
    """
    return self._flow_manager.StopCapturingAudio(port_id)

  def StartPlayingAudio(self, port_id, path, data_format):
    """Playing audio data from an output port.

    Play audio data at given path using given format from port_id port.

    Args:
      port_id: The ID of the output connector.
      path: The path to the audio data to play.
      data_format: The dict representation of AudioDataFormat.
        Refer to docstring of utils.audio.AudioDataFormat for detail.
        Currently Chameleon only accepts data format if it meets
        dict(file_type='raw', sample_format='S32_LE', channel=8, rate=48000)
        Chameleon user should do the format conversion to minimize work load
        on Chameleon board.
    """
    self._flow_manager.StartPlayingAudio(port_id, path, data_format)

  def StartPlayingEcho(self, port_id, input_id):
    """Echoes audio data received from input_id and plays to port_id.

    Echoes audio data received from input_id and plays to port_id.

    Chameleon does not support echoing from HDMI and capturing from LineIn/Mic
    at the same time. The echoing/capturing needs to be stop first before
    another action starts.

    For example, user can call

    StartPlayingEcho(3, 7) --> StopPlayingAudio(3) --> StartCapturingAudio(6)

    or

    StartCapturingAudio(6) --> StopCapturingAudio(6) --> StartPlayingEcho(3, 7)

    but user can not call

    StartPlayingEcho(3, 7) --> StartCapturingAudio(6)

    or

    StartCapturingAudio(6) --> StartPlayingEcho(3, 7)

    Exception is raised when conflicting actions are performed.

    Args:
      port_id: The ID of the output connector. Check the value in ids.py.
      input_id: The ID of the input connector. Check the value in ids.py.
    """
    self._flow_manager.StartPlayingEcho(port_id, input_id)

  def StopPlayingAudio(self, port_id):
    """Stops playing audio from port_id port.

    Args:
      port_id: The ID of the output connector.
    """
    self._flow_manager.StopPlayingAudio(port_id)

  def _ClearPrinterFiles(self):
    """Clears temporary printer files.

    Chameleon board does not reboot very often. We should clear the temporary
    printer files used in capturing printer data.
    """
    for path in glob.glob('/tmp/printer_*'):
      os.unlink(path)

  def _ClearAudioFiles(self):
    """Clears temporary audio files.

    Chameleon board does not reboot very often. We should clear the temporary
    audio files used in capturing audio or playing audio when Reset is called.
    """
    for path in glob.glob('/tmp/audio_*'):
      os.unlink(path)

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardConnect(self, bus_number, endpoint):
    """Connects an endpoint to an audio bus.

    Args:
      bus_number: 1 or 2 for audio bus 1 or bus 2.
      endpoint: An endpoint defined in audio_board.AudioBusEndpoint.

    Raises:
      DriverError: If the endpoint is a source and there is other source
                   endpoint occupying audio bus.
    """
    if audio_board.IsSource(endpoint):
      current_sources, _ = self.audio_board.GetConnections(bus_number)
      if current_sources and endpoint not in current_sources:
        raise DriverError(
            'Sources %s other than %s are currently occupying audio bus.' %
            (current_sources, endpoint))

    self.audio_board.SetConnection(
        bus_number, endpoint, True)

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardDisconnect(self, bus_number, endpoint):
    """Disconnects an endpoint to an audio bus.

    Args:
      bus_number: 1 or 2 for audio bus 1 or bus 2.
      endpoint: An endpoint defined in audio_board.AudioBusEndpoint.

    Raises:
      DriverError: If the endpoint is not connected to audio bus.
    """
    if not self.audio_board.IsConnected(bus_number, endpoint):
      raise DriverError(
          'Endpoint %s is not connected to audio bus %d.' %
          (endpoint, bus_number))

    self.audio_board.SetConnection(
        bus_number, endpoint, False)

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardGetRoutes(self, bus_number):
    """Gets a list of routes on audio bus.

    Args:
      bus_number: 1 or 2 for audio bus 1 or bus 2.

    Returns:
      A list of tuples (source, sink) that are routed on audio bus
      where source and sink are endpoints defined in
      audio_board.AudioBusEndpoint.
    """
    sources, sinks = self.audio_board.GetConnections(bus_number)
    routes = []
    for source in sources:
      for sink in sinks:
        logging.info('Route on bus %d: %s ---> %s',
                     bus_number, source, sink)
        routes.append((source, sink))
    return routes

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardClearRoutes(self, bus_number):
    """Clears routes on an audio bus.

    Args:
      bus_number: 1 or 2 for audio bus 1 or bus 2.
    """
    self.audio_board.ResetConnections(bus_number)

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardHasJackPlugger(self):
    """If there is jack plugger on audio board.

    Audio board must have the motor cable connected in order to control
    jack plugger of audio box.

    Returns:
      True if there is jack plugger on audio board. False otherwise.
    """
    return self.audio_board.HasJackPlugger()

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardAudioJackPlug(self):
    """Plugs audio jack to connect audio board and Cros device."""
    logging.info('Plug audio jack to connect audio board and Cros device.')
    self.audio_board.SetJackPlugger(True)

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardAudioJackUnplug(self):
    """Unplugs audio jack to disconnect audio board and Cros device."""
    logging.info('Unplug audio jack to disconnect audio board and Cros device.')
    self.audio_board.SetJackPlugger(False)

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardResetBluetooth(self):
    """Resets bluetooth module on audio board."""
    logging.info('Resets bluetooth module on audio board.')
    self.audio_board.ResetBluetooth()

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardDisableBluetooth(self):
    """Disables bluetooth module on audio board."""
    logging.info('Disables bluetooth module on audio board.')
    self.audio_board.DisableBluetooth()

  @_DeviceMethod(ids.AUDIO_BOARD)
  def AudioBoardIsBluetoothEnabled(self):
    """Checks if bluetooth module on audio board is enabled.

    Returns:
      True if bluetooth module is enabled. False otherwise.
    """
    return self.audio_board.IsBluetoothEnabled()

  def SetUSBDriverPlaybackConfigs(self, playback_data_format):
    """Updates the corresponding playback configurations to argument values.

    This provides flexibility for simulating the USB gadget driver using other
    configurations different from the default values.

    Args:
      playback_data_format: The dict form of an AudioDataFormat object. The
        'file_type' field will be ignored since for playback, there is no need
        for setting file type before playing audio. It is specified by the audio
        file passed in for playback. Other fields are used to set USB driver
        configurations.
    """
    self._flow_manager.SetUSBDriverPlaybackConfigs(playback_data_format)

  def SetUSBDriverCaptureConfigs(self, capture_data_format):
    """Updates the corresponding capture configurations to argument values.

    This provides flexibility for simulating the USB gadget driver using other
    configurations different from the default values.

    Args:
      capture_data_format: The dict form of an AudioDataFormat object. The
        'file_type' field will be saved by InputUSBAudioFlow as the file type
        for captured data. Other fields are used to set USB driver
        configurations.
    """
    self._flow_manager.SetUSBDriverCaptureConfigs(capture_data_format)

  def GetMacAddress(self):
    """Gets the MAC address of this Chameleon.

    Returns:
      A string for MAC address.
    """
    return open('/sys/class/net/eth0/address').read().strip()

  def SendHIDEvent(self, port_id, event_type, *args, **kwargs):
    """Sends HID event with event_type and arguments for HID port #port_id.

    Args:
      port_id: The ID of the HID port.
      event_type: Supported event type of string for HID port #port_id.

    Returns:
      Returns as event function if applicable.
    """
    return self._flow_manager.SendHIDEvent(port_id, event_type, *args, **kwargs)

  def ResetBluetoothRef(self):
    """Reset BTREF"""
    # Reset the RN52 and wait for it to reset
    RN52_RESET_WAIT_SECS = 3
    self.bluetooth_a2dp_sink.PowerCycle()
    time.sleep(RN52_RESET_WAIT_SECS)
    # Reloads serial port driver if needed
    self._bluetooth_a2dp_sink_ctrl.EnableDriver()
    # Reads peripheral configuration settings
    self.bluetooth_a2dp_sink.GetBasicSettings()

  def EnableBluetoothRef(self):
    """Enable BTREF"""
    self._bluetooth_a2dp_sink_ctrl.EnableDriver()

  def DisableBluetoothRef(self):
    """Disable BTREF"""
    self.bluetooth_a2dp_sink.Close()

  def IsBluetoothRefDisabled(self):
    """Check if BTREF is enabled"""
    raise NotImplementedError('IsBluetoothRefDisabled')