diff options
Diffstat (limited to 'mmi2grpc/a2dp.py')
| -rw-r--r-- | mmi2grpc/a2dp.py | 590 |
1 files changed, 590 insertions, 0 deletions
diff --git a/mmi2grpc/a2dp.py b/mmi2grpc/a2dp.py new file mode 100644 index 0000000..856e45d --- /dev/null +++ b/mmi2grpc/a2dp.py @@ -0,0 +1,590 @@ +# 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" |