xref: /external/autotest/client/cros/chameleon/chameleon_audio_helper.py

# Lint as: python2, python3
# Copyright 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""This module provides the framework for audio tests using chameleon."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import logging
from contextlib import contextmanager

from autotest_lib.client.cros.chameleon import audio_widget
from autotest_lib.client.cros.chameleon import audio_widget_arc
from autotest_lib.client.cros.chameleon import audio_widget_link
from autotest_lib.server.cros.bluetooth import bluetooth_device
from autotest_lib.client.cros.chameleon import chameleon_audio_ids as ids
import six
from six.moves import range

class AudioPort(object):
    """
    This class abstracts an audio port in audio test framework. A port is
    identified by its host and interface. Available hosts and interfaces
    are listed in chameleon_audio_ids.

    Properties:
        port_id: The port id defined in chameleon_audio_ids.
        host: The host of this audio port, e.g. 'Chameleon', 'Cros',
              'Peripheral'.
        interface: The interface of this audio port, e.g. 'HDMI', 'Headphone'.
        role: The role of this audio port, that is, 'source' or
              'sink'. Note that bidirectional interface like 3.5mm
              jack is separated to two interfaces 'Headphone' and
             'External Mic'.

    """
    def __init__(self, port_id):
        """Initialize an AudioPort with port id string.

        @param port_id: A port id string defined in chameleon_audio_ids.

        """
        logging.debug('Creating AudioPort with port_id: %s', port_id)
        self.port_id = port_id
        self.host = ids.get_host(port_id)
        self.interface = ids.get_interface(port_id)
        self.role = ids.get_role(port_id)
        logging.debug('Created AudioPort: %s', self)


    def __str__(self):
        """String representation of audio port.

        @returns: The string representation of audio port which is composed by
                  host, interface, and role.

        """
        return '( %s | %s | %s )' % (
                self.host, self.interface, self.role)


class AudioLinkFactoryError(Exception):
    """Error in AudioLinkFactory."""
    pass


class AudioLinkFactory(object):
    """
    This class provides method to create link that connects widgets.
    This is used by AudioWidgetFactory when user wants to create binder for
    widgets.

    Properties:
        _audio_bus_links: A dict containing mapping from index number
                          to object of AudioBusLink's subclass.
        _audio_board: An AudioBoard object to access Chameleon
                      audio board functionality.

    """

    # Maps pair of widgets to widget link of different type.
    LINK_TABLE = {
        (ids.CrosIds.HDMI, ids.ChameleonIds.HDMI):
                audio_widget_link.HDMIWidgetLink,
        (ids.CrosIds.HEADPHONE, ids.ChameleonIds.LINEIN):
                audio_widget_link.AudioBusToChameleonLink,
        (ids.ChameleonIds.LINEOUT, ids.CrosIds.EXTERNAL_MIC):
                audio_widget_link.AudioBusToCrosLink,
        (ids.ChameleonIds.LINEOUT, ids.PeripheralIds.SPEAKER):
                audio_widget_link.AudioBusChameleonToPeripheralLink,
        (ids.PeripheralIds.MIC, ids.ChameleonIds.LINEIN):
                audio_widget_link.AudioBusToChameleonLink,
        (ids.PeripheralIds.BLUETOOTH_DATA_RX,
         ids.ChameleonIds.LINEIN):
                audio_widget_link.AudioBusToChameleonLink,
        (ids.ChameleonIds.LINEOUT,
         ids.PeripheralIds.BLUETOOTH_DATA_TX):
                audio_widget_link.AudioBusChameleonToPeripheralLink,
        (ids.CrosIds.BLUETOOTH_HEADPHONE,
         ids.PeripheralIds.BLUETOOTH_DATA_RX):
                audio_widget_link.BluetoothHeadphoneWidgetLink,
        (ids.PeripheralIds.BLUETOOTH_DATA_TX,
         ids.CrosIds.BLUETOOTH_MIC):
                audio_widget_link.BluetoothMicWidgetLink,
        (ids.CrosIds.USBOUT, ids.ChameleonIds.USBIN):
                audio_widget_link.USBToChameleonWidgetLink,
        (ids.ChameleonIds.USBOUT, ids.CrosIds.USBIN):
                audio_widget_link.USBToCrosWidgetLink,
        # TODO(cychiang): Add link for other widget pairs.
    }

    def __init__(self, cros_host):
        """Initializes an AudioLinkFactory.

        @param cros_host: A CrosHost object to access Cros device.

        """
        # There are two audio buses on audio board. Initializes these links
        # to None. They may be changed to objects of AudioBusLink's subclass.
        self._audio_bus_links = {1: None, 2: None}
        self._cros_host = cros_host
        self._chameleon_board = cros_host.chameleon
        self._audio_board = self._chameleon_board.get_audio_board()
        self._bluetooth_device = None
        self._usb_ctrl = None


    def _acquire_audio_bus_index(self):
        """Acquires an available audio bus index that is not occupied yet.

        @returns: A number.

        @raises: AudioLinkFactoryError if there is no available
                 audio bus.
        """
        for index, bus in six.iteritems(self._audio_bus_links):
            if not (bus and bus.occupied):
                return index

        raise AudioLinkFactoryError('No available audio bus')


    def create_link(self, source, sink):
        """Creates a widget link for two audio widgets.

        @param source: An AudioWidget.
        @param sink: An AudioWidget.

        @returns: An object of WidgetLink's subclass.

        @raises: AudioLinkFactoryError if there is no link between
            source and sink.

        """
        # Finds the available link types from LINK_TABLE.
        link_type = self.LINK_TABLE.get((source.port_id, sink.port_id), None)
        if not link_type:
            raise AudioLinkFactoryError(
                    'No supported link between %s and %s' % (
                            source.port_id, sink.port_id))

        # There is only one dedicated HDMI cable, just use it.
        if link_type == audio_widget_link.HDMIWidgetLink:
            link = audio_widget_link.HDMIWidgetLink(self._cros_host)

        # Acquires audio bus if there is available bus.
        # Creates a bus of AudioBusLink's subclass that is more
        # specific than AudioBusLink.
        # Controls this link using AudioBus object obtained from AudioBoard
        # object.
        elif issubclass(link_type, audio_widget_link.AudioBusLink):
            bus_index = self._acquire_audio_bus_index()
            link = link_type(self._audio_board.get_audio_bus(bus_index))
            self._audio_bus_links[bus_index] = link
        elif issubclass(link_type, audio_widget_link.BluetoothWidgetLink):
            # To connect bluetooth adapter on Cros device to bluetooth module on
            # chameleon board, we need to access bluetooth adapter on Cros host
            # using BluetoothDevice, and access bluetooth module on
            # audio board using BluetoothController.

            # Initializes a BluetoothDevice object if needed. And reuse this
            # object for future bluetooth link usage.
            if not self._bluetooth_device:
                self._bluetooth_device = bluetooth_device.BluetoothDevice(
                        self._cros_host)
            link = link_type(
                    self._bluetooth_device,
                    self._chameleon_board.get_bluetooth_ref_controller(),
                    self._chameleon_board.get_bluetooth_a2dp_sink().GetLocalBluetoothAddress())
        elif issubclass(link_type, audio_widget_link.USBWidgetLink):
            # Aside from managing connection between USB audio gadget driver on
            # Chameleon with Cros device, USBWidgetLink also handles changing
            # the gadget driver's configurations, through the USBController that
            # is passed to it at initialization.
            if not self._usb_ctrl:
                self._usb_ctrl = self._chameleon_board.get_usb_controller()

            link = link_type(self._usb_ctrl)
        else:
            raise NotImplementedError('Link %s is not implemented' % link_type)

        return link


class AudioWidgetFactoryError(Exception):
    """Error in AudioWidgetFactory."""
    pass


class AudioWidgetFactory(object):
    """
    This class provides methods to create widgets and binder of widgets.
    User can use binder to setup audio paths. User can use widgets to control
    record/playback on different ports on Cros device or Chameleon.

    Properties:
        _audio_facade: An AudioFacadeRemoteAdapter to access Cros device audio
                       functionality. This is created by the
                       'factory' argument passed to the constructor.
        _display_facade: A DisplayFacadeRemoteAdapter to access Cros device
                         display functionality. This is created by the
                         'factory' argument passed to the constructor.
        _system_facade: A SystemFacadeRemoteAdapter to access Cros device
                         system functionality. This is created by the
                         'factory' argument passed to the constructor.
        _chameleon_board: A ChameleonBoard object to access Chameleon
                          functionality.
        _link_factory: An AudioLinkFactory that creates link for widgets.

    """
    def __init__(self, factory, cros_host):
        """Initializes a AudioWidgetFactory

        @param factory: A facade factory to access Cros device functionality.
                        Currently only audio facade is used, but we can access
                        other functionalities including display and video by
                        facades created by this facade factory.
        @param cros_host: A CrosHost object to access Cros device.

        """
        self._audio_facade = factory.create_audio_facade()
        self._display_facade = factory.create_display_facade()
        self._system_facade = factory.create_system_facade()
        self._usb_facade = factory.create_usb_facade()
        self._cros_host = cros_host
        self._chameleon_board = cros_host.chameleon
        self._link_factory = AudioLinkFactory(cros_host)


    def create_widget(self, port_id, use_arc=False):
        """Creates a AudioWidget given port id string.

        @param port_id: A port id string defined in chameleon_audio_ids.
        @param use_arc: For Cros widget, select if audio path exercises ARC.
                        Currently only input widget is supported.

        @returns: An AudioWidget that is actually a
                  (Chameleon/Cros/Peripheral)(Input/Output)Widget.

        """
        def _create_chameleon_handler(audio_port):
            """Creates a ChameleonWidgetHandler for a given AudioPort.

            @param audio_port: An AudioPort object.

            @returns: A Chameleon(Input/Output)WidgetHandler depending on
                      role of audio_port.

            """
            if audio_port.role == 'sink':
                if audio_port.port_id == ids.ChameleonIds.HDMI:
                    return audio_widget.ChameleonHDMIInputWidgetHandler(
                            self._chameleon_board, audio_port.interface,
                            self._display_facade)
                else:
                    return audio_widget.ChameleonInputWidgetHandler(
                            self._chameleon_board, audio_port.interface)
            else:
                if audio_port.port_id == ids.ChameleonIds.LINEOUT:
                    return audio_widget.ChameleonLineOutOutputWidgetHandler(
                            self._chameleon_board, audio_port.interface)
                else:
                    return audio_widget.ChameleonOutputWidgetHandler(
                            self._chameleon_board, audio_port.interface)

        def _create_cros_handler(audio_port):
            """Creates a CrosWidgetHandler for a given AudioPort.

            @param audio_port: An AudioPort object.

            @returns: A Cros(Input/Output)WidgetHandler depending on
                      role of audio_port.

            """
            is_usb = audio_port.port_id in [
                    ids.CrosIds.USBIN, ids.CrosIds.USBOUT
            ]
            is_internal_mic = audio_port.port_id == ids.CrosIds.INTERNAL_MIC
            is_hotwording = audio_port.port_id == ids.CrosIds.HOTWORDING

            if audio_port.role == 'sink':
                if use_arc:
                    return audio_widget_arc.CrosInputWidgetARCHandler(
                            self._audio_facade)
                elif is_usb:
                    return audio_widget.CrosUSBInputWidgetHandler(
                            self._audio_facade)
                elif is_hotwording:
                    return audio_widget.CrosHotwordingWidgetHandler(
                            self._audio_facade, self._system_facade)
                else:
                    return audio_widget.CrosInputWidgetHandler(
                            self._audio_facade)
            else:
                if use_arc:
                    return audio_widget_arc.CrosOutputWidgetARCHandler(
                            self._audio_facade)
                return audio_widget.CrosOutputWidgetHandler(
                        self._audio_facade)

        def _create_audio_widget(audio_port, handler):
            """Creates an AudioWidget for given AudioPort using WidgetHandler.

            Creates an AudioWidget with the role of audio_port. Put
            the widget handler into the widget so the widget can handle
            action requests.

            @param audio_port: An AudioPort object.
            @param handler: A WidgetHandler object.

            @returns: An Audio(Input/Output)Widget depending on
                      role of audio_port.

            @raises: AudioWidgetFactoryError if fail to create widget.

            """
            if audio_port.host in ['Chameleon', 'Cros']:
                if audio_port.role == 'sink':
                    return audio_widget.AudioInputWidget(audio_port, handler)
                else:
                    return audio_widget.AudioOutputWidget(audio_port, handler)
            elif audio_port.host == 'Peripheral':
                return audio_widget.PeripheralWidget(audio_port, handler)
            else:
                raise AudioWidgetFactoryError(
                        'The host %s is not valid' % audio_port.host)


        audio_port = AudioPort(port_id)
        if audio_port.host == 'Chameleon':
            handler = _create_chameleon_handler(audio_port)
        elif audio_port.host == 'Cros':
            handler = _create_cros_handler(audio_port)
        elif audio_port.host == 'Peripheral':
            handler = audio_widget.PeripheralWidgetHandler()

        return _create_audio_widget(audio_port, handler)


    def _create_widget_binder(self, source, sink):
        """Creates a WidgetBinder for two AudioWidgets.

        @param source: An AudioWidget.
        @param sink: An AudioWidget.

        @returns: A WidgetBinder object.

        """
        return audio_widget_link.WidgetBinder(
                source, self._link_factory.create_link(source, sink), sink)


    def create_binder(self, *widgets):
        """Creates a WidgetBinder or a WidgetChainBinder for AudioWidgets.

        @param widgets: A list of widgets that should be linked in a chain.

        @returns: A WidgetBinder for two widgets. A WidgetBinderChain object
                  for three or more widgets.

        """
        if len(widgets) == 2:
            return self._create_widget_binder(widgets[0], widgets[1])
        binders = []
        for index in range(len(widgets) - 1):
            binders.append(
                    self._create_widget_binder(
                            widgets[index],  widgets[index + 1]))

        return audio_widget_link.WidgetBinderChain(binders)


@contextmanager
def bind_widgets(binder):
    """Context manager for widget binders.

    Connects widgets in the beginning. Disconnects widgets and releases binder
    in the end.

    @param binder: A WidgetBinder object or a WidgetBinderChain object.
                   If binder is None, then do nothing. This is for test user's
                   convenience to reuse test logic among paths using binder
                   and paths not using binder.

    E.g. with bind_widgets(binder):
             do something on widget.

    """
    if not binder:
        yield
    else:
        try:
            binder.connect()
            yield
        finally:
            binder.disconnect()
            binder.release()