path: root/mmi2grpc/a2dp.py

# Copyright 2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""A2DP proxy module."""

import time
from typing import Optional

from grpc import RpcError

from mmi2grpc._audio import AudioSignal
from mmi2grpc._helpers import assert_description
from mmi2grpc._proxy import ProfileProxy
from pandora.a2dp_grpc import A2DP
from pandora.a2dp_pb2 import Sink, Source, PlaybackAudioRequest
from pandora.host_grpc import Host
from pandora.host_pb2 import Connection

AUDIO_SIGNAL_AMPLITUDE = 0.8
AUDIO_SIGNAL_SAMPLING_RATE = 44100


class A2DPProxy(ProfileProxy):
    """A2DP proxy.

    Implements A2DP and AVDTP PTS MMIs.
    """

    connection: Optional[Connection] = None
    sink: Optional[Sink] = None
    source: Optional[Source] = None

    def __init__(self, channel):
        super().__init__()

        self.host = Host(channel)
        self.a2dp = A2DP(channel)

        def convert_frame(data):
            return PlaybackAudioRequest(data=data, source=self.source)
        self.audio = AudioSignal(
            lambda frames: self.a2dp.PlaybackAudio(map(convert_frame, frames)),
            AUDIO_SIGNAL_AMPLITUDE,
            AUDIO_SIGNAL_SAMPLING_RATE
        )

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_connect(
            self, test: str, pts_addr: bytes, **kwargs):
        """
        If necessary, take action to accept the AVDTP Signaling Channel
        Connection initiated by the tester.

        Description: Make sure the IUT
        (Implementation Under Test) is in a state to accept incoming Bluetooth
        connections.  Some devices may need to be on a specific screen, like a
        Bluetooth settings screen, in order to pair with PTS.  If the IUT is
        still having problems pairing with PTS, try running a test case where
        the IUT connects to PTS to establish pairing.
        """

        if "SRC" in test:
            self.connection = self.host.WaitConnection(
                address=pts_addr).connection
            try:
                if "INT" in test:
                    self.source = self.a2dp.OpenSource(
                        connection=self.connection).source
                else:
                    self.source = self.a2dp.WaitSource(
                        connection=self.connection).source
            except RpcError:
                pass
        else:
            self.connection = self.host.WaitConnection(
                address=pts_addr).connection
            try:
                self.sink = self.a2dp.WaitSink(
                    connection=self.connection).sink
            except RpcError:
                pass
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_discover(self, **kwargs):
        """
        Send a discover command to PTS.

        Action: If the IUT (Implementation
        Under Test) is already connected to PTS, attempting to send or receive
        streaming media should trigger this action.  If the IUT is not connected
        to PTS, attempting to connect may trigger this action.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_start(self, test: str, **kwargs):
        """
        Send a start command to PTS.

        Action: If the IUT (Implementation Under
        Test) is already connected to PTS, attempting to send or receive
        streaming media should trigger this action.  If the IUT is not connected
        to PTS, attempting to connect may trigger this action.
        """

        if "SRC" in test:
            self.a2dp.Start(source=self.source)
        else:
            self.a2dp.Start(sink=self.sink)
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_suspend(self, test: str, **kwargs):
        """
        Suspend the streaming channel.
        """

        if "SRC" in test:
            self.a2dp.Suspend(source=self.source)
        else:
            assert False
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_close_stream(self, test: str, **kwargs):
        """
        Close the streaming channel.

        Action: Disconnect the streaming channel,
        or close the Bluetooth connection to the PTS.
        """

        if "SRC" in test:
            self.a2dp.Close(source=self.source)
            self.source = None
        else:
            self.a2dp.Close(sink=self.sink)
            self.sink = None
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_out_of_range(
            self, pts_addr: bytes, **kwargs):
        """
        Move the IUT out of range to create a link loss scenario.

        Action: This
        can be also be done by placing the IUT or PTS in an RF shielded box.
         """

        if self.connection is None:
            self.connection = self.host.GetConnection(
                address=pts_addr).connection
        self.host.Disconnect(connection=self.connection)
        self.connection = None
        self.sink = None
        self.source = None
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_begin_streaming(self, test: str, **kwargs):
        """
        Begin streaming media ...

        Note: If the IUT has suspended the stream
        please restart the stream to begin streaming media.
        """

        if test == "AVDTP/SRC/ACP/SIG/SMG/BI-29-C":
            time.sleep(2)  # TODO: Remove, AVRCP SegFault
        if test in ("A2DP/SRC/CC/BV-09-I",
                    "A2DP/SRC/SET/BV-04-I",
                    "AVDTP/SRC/ACP/SIG/SMG/BV-18-C",
                    "AVDTP/SRC/ACP/SIG/SMG/BV-20-C",
                    "AVDTP/SRC/ACP/SIG/SMG/BV-22-C"):
            time.sleep(1)  # TODO: Remove, AVRCP SegFault
        if test == "A2DP/SRC/SUS/BV-01-I":
            # Stream is not suspended when we receive the interaction
            time.sleep(1)

        self.a2dp.Start(source=self.source)
        self.audio.start()
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_media(self, **kwargs):
        """
        Take action if necessary to start streaming media to the tester.
        """

        self.audio.start()
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_stream_media(self, **kwargs):
        """
        Stream media to PTS.  If the IUT is a SNK, wait for PTS to start
        streaming media.

        Action: If the IUT (Implementation Under Test) is
        already connected to PTS, attempting to send or receive streaming media
        should trigger this action.  If the IUT is not connected to PTS,
        attempting to connect may trigger this action.
        """

        self.audio.start()
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_user_verify_media_playback(self, **kwargs):
        """
        Is the test system properly playing back the media being sent by the
        IUT?
        """

        result = self.audio.verify()
        assert result

        return "Yes" if result else "No"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_get_capabilities(self, **kwargs):
        """
        Send a get capabilities command to PTS.

        Action: If the IUT
        (Implementation Under Test) is already connected to PTS, attempting to
        send or receive streaming media should trigger this action.  If the IUT
        is not connected to PTS, attempting to connect may trigger this action.
        """

        # This will be done as part as the a2dp.OpenSource or a2dp.WaitSource
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_discover(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Discover operation
        initiated by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_set_configuration(self, **kwargs):
        """
        Send a set configuration command to PTS.

        Action: If the IUT
        (Implementation Under Test) is already connected to PTS, attempting to
        send or receive streaming media should trigger this action.  If the IUT
        is not connected to PTS, attempting to connect may trigger this action.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_close_stream(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Close operation initiated
        by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_abort(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Abort operation initiated
        by the tester..
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_get_all_capabilities(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Get All Capabilities
        operation initiated by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_get_capabilities(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Get Capabilities operation
        initiated by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_set_configuration(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Set Configuration
        operation initiated by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_get_configuration(self, **kwargs):
        """
        Take action to accept the AVDTP Get Configuration command from the
        tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_open_stream(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Open operation initiated
        by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_start(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Start operation initiated
        by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_suspend(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Suspend operation
        initiated by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_reconfigure(self, **kwargs):
        """
        If necessary, take action to accept the AVDTP Reconfigure operation
        initiated by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_media_transports(self, **kwargs):
        """
        Take action to accept transport channels for the recently configured
        media stream.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_confirm_streaming(self, **kwargs):
        """
        Is the IUT (Implementation Under Test) receiving streaming media from
        PTS?

        Action: Press 'Yes' if the IUT is receiving streaming data from
        the PTS (in some cases the sound may not be clear, this is normal).
        """

        # TODO: verify
        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_open_stream(self, **kwargs):
        """
        Open a streaming media channel.

        Action: If the IUT (Implementation
        Under Test) is already connected to PTS, attempting to send or receive
        streaming media should trigger this action.  If the IUT is not connected
        to PTS, attempting to connect may trigger this action.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_reconnect(self, pts_addr: bytes, **kwargs):
        """
        Press OK when the IUT (Implementation Under Test) is ready to allow the
        PTS to reconnect the AVDTP signaling channel.

        Action: Press OK when the
        IUT is ready to accept Bluetooth connections again.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_iut_initiate_get_all_capabilities(self, **kwargs):
        """
        Send a GET ALL CAPABILITIES command to PTS.

        Action: If the IUT
        (Implementation Under Test) is already connected to PTS, attempting to
        send or receive streaming media should trigger this action.  If the IUT
        is not connected to PTS, attempting to connect may trigger this action.
        """

        return "OK"

    @assert_description
    def TSC_AVDTP_mmi_tester_verifying_suspend(self, **kwargs):
        """
        Please wait while the tester verifies the IUT does not send media during
        suspend ...
        """

        return "Yes"

    @assert_description
    def TSC_A2DP_mmi_user_confirm_optional_data_attribute(self, **kwargs):
        """
        Tester found the optional SDP attribute named 'Supported Features'.
        Press 'Yes' if the data displayed below is correct.

        Value: 0x0001
        """

        # TODO: Extract and verify attribute name and value from description
        return "OK"

    @assert_description
    def TSC_A2DP_mmi_user_confirm_optional_string_attribute(self, **kwargs):
        """
        Tester found the optional SDP attribute named 'Service Name'.  Press
        'Yes' if the string displayed below is correct.

        Value: Advanced Audio
        Source
        """

        # TODO: Extract and verify attribute name and value from description
        return "OK"

    @assert_description
    def TSC_A2DP_mmi_user_confirm_no_optional_attribute_support(self, **kwargs):
        """
        Tester could not find the optional SDP attribute named 'Provider Name'.
        Is this correct?
        """

        # TODO: Extract and verify attribute name from description
        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_accept_delayreport(self, **kwargs):
        """
        Take action if necessary to accept the Delay Reportl command from the
        tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_initiate_media_transport_connect(self, **kwargs):
        """
        Take action to initiate an AVDTP media transport.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_user_confirm_SIG_SMG_BV_28_C(self, **kwargs):
        """
        Were all the service capabilities reported to the upper tester valid?
        """

        # TODO: verify
        return "Yes"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_reject_invalid_command(self, **kwargs):
        """
        Take action to reject the invalid command sent by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_reject_open(self, **kwargs):
        """
        Take action to reject the invalid OPEN command sent by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_reject_start(self, **kwargs):
        """
        Take action to reject the invalid START command sent by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_reject_suspend(self, **kwargs):
        """
        Take action to reject the invalid SUSPEND command sent by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_reject_reconfigure(self, **kwargs):
        """
        Take action to reject the invalid or incompatible RECONFIGURE command
        sent by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_reject_get_all_capabilities(self, **kwargs):
        """
        Take action to reject the invalid GET ALL CAPABILITIES command with the
        error code BAD_LENGTH.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_reject_get_capabilities(self, **kwargs):
        """
        Take action to reject the invalid GET CAPABILITIES command with the
        error code BAD_LENGTH.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_reject_set_configuration(self, **kwargs):
        """
        Take action to reject the SET CONFIGURATION sent by the tester.  The IUT
        is expected to respond with SEP_IN_USE because the SEP requested was
        previously configured.
        """

        return "OK"

    def TSC_AVDTPEX_mmi_iut_reject_get_configuration(self, **kwargs):
        """
        Take action to reject the GET CONFIGURATION sent by the tester.  The IUT
        is expected to respond with BAD_ACP_SEID because the SEID requested was
        not previously configured.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_iut_reject_close(self, **kwargs):
        """
        Take action to reject the invalid CLOSE command sent by the tester.
        """

        return "OK"

    @assert_description
    def TSC_AVDTPEX_mmi_user_confirm_SIG_SMG_BV_18_C(self, **kwargs):
        """
        Did the IUT receive media with the following information?

        - V = RTP_Ver
        - P = 0 (no padding bits)
        - X = 0 (no extension)
        - CC = 0 (no
        contributing source)
        - M = 0
        """

        # TODO: verify
        return "OK"