Merge upstream commit '6029c0575763ef237440a9bb7447a922eecf47fb' into master am: 8b4d56f505 am: 2fe5b84081 am: 440eae6d5f am: 1f7cc85b8e am: b64f9712e1
Original change: https://android-review.googlesource.com/c/platform/external/pandora/mmi2grpc/+/2168183
Change-Id: I20fcb7aeac53388ea4ca9a7f28a18b14bb0a818f
Signed-off-by: Automerger Merge Worker <android-build-automerger-merge-worker@system.gserviceaccount.com>
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..bee8a64
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+__pycache__
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 0000000..97c24f3
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,30 @@
+# How to Contribute
+
+We'd love to accept your patches and contributions to this project. There are
+just a few small guidelines you need to follow.
+
+## Contributor License Agreement
+
+Contributions to this project must be accompanied by a Contributor License
+Agreement. You (or your employer) retain the copyright to your contribution;
+this simply gives us permission to use and redistribute your contributions as
+part of the project. Head over to <https://cla.developers.google.com/> to see
+your current agreements on file or to sign a new one.
+
+You generally only need to submit a CLA once, so if you've already submitted one
+(even if it was for a different project), you probably don't need to do it
+again.
+
+## Style Guide
+
+Every contributions must follow [Google Python style guide](
+https://google.github.io/styleguide/pyguide.html).
+
+## Code Reviews
+
+All submissions, including submissions by project members, require review.
+
+## Community Guidelines
+
+This project follows [Google's Open Source Community
+Guidelines](https://opensource.google/conduct/).
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..d645695
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,202 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ 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
+
+ http://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.
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..c9fec3a
--- /dev/null
+++ b/README.md
@@ -0,0 +1,17 @@
+# mmi2grpc
+
+## Install
+
+```bash
+git submodule update --init
+
+pip install [-e] bt-test-interfaces/python
+pip install [-e] .
+```
+
+## Rebuild gRPC Bluetooth test interfaces
+
+```bash
+pip install grpcio-tools==1.46.3
+./bt-test-interfaces/python/_build/grpc.py
+```
diff --git a/__init__.py b/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/__init__.py
diff --git a/mmi2grpc/__init__.py b/mmi2grpc/__init__.py
new file mode 100644
index 0000000..f897f46
--- /dev/null
+++ b/mmi2grpc/__init__.py
@@ -0,0 +1,118 @@
+# 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.
+
+"""Map Bluetooth PTS Man Machine Interface to Pandora gRPC calls."""
+
+__version__ = "0.0.1"
+
+from typing import List
+import time
+import sys
+
+import grpc
+
+from mmi2grpc.a2dp import A2DPProxy
+from mmi2grpc._helpers import format_proxy
+from pandora.host_grpc import Host
+
+GRPC_PORT = 8999
+MAX_RETRIES = 10
+
+
+class IUT:
+ """IUT class.
+
+ Handles MMI calls from the PTS and routes them to corresponding profile
+ proxy which translates MMI calls to gRPC calls to the IUT.
+ """
+ def __init__(
+ self, test: str, args: List[str], port: int = GRPC_PORT, **kwargs):
+ """Init IUT class for a given test.
+
+ Args:
+ test: PTS test id.
+ args: test arguments.
+ port: gRPC port exposed by the IUT test server.
+ """
+ self.port = port
+ self.test = test
+
+ # Profile proxies.
+ self._a2dp = None
+
+ def __enter__(self):
+ """Resets the IUT when starting a PTS test."""
+ # Note: we don't keep a single gRPC channel instance in the IUT class
+ # because reset is allowed to close the gRPC server.
+ with grpc.insecure_channel(f'localhost:{self.port}') as channel:
+ Host(channel).Reset(wait_for_ready=True)
+
+ def __exit__(self, exc_type, exc_value, exc_traceback):
+ self._a2dp = None
+
+ @property
+ def address(self) -> bytes:
+ """Bluetooth MAC address of the IUT."""
+ with grpc.insecure_channel(f'localhost:{self.port}') as channel:
+ tries = 0
+ while True:
+ try:
+ return Host(channel).ReadLocalAddress(
+ wait_for_ready=True).address
+ except grpc.RpcError or grpc._channel._InactiveRpcError:
+ tries += 1
+ if tries >= MAX_RETRIES:
+ raise
+ else:
+ print('Retry', tries, 'of', MAX_RETRIES)
+ time.sleep(1)
+
+ def interact(self,
+ pts_address: bytes,
+ profile: str,
+ test: str,
+ interaction: str,
+ description: str,
+ style: str,
+ **kwargs) -> str:
+ """Routes MMI calls to corresponding profile proxy.
+
+ Args:
+ pts_address: Bluetooth MAC addres of the PTS in bytes.
+ profile: Bluetooth profile.
+ test: PTS test id.
+ interaction: MMI name.
+ description: MMI description.
+ style: MMI popup style, unused for now.
+ """
+ print(f'{profile} mmi: {interaction}', file=sys.stderr)
+
+ # Handles A2DP and AVDTP MMIs.
+ if profile in ('A2DP', 'AVDTP'):
+ if not self._a2dp:
+ self._a2dp = A2DPProxy(
+ grpc.insecure_channel(f'localhost:{self.port}'))
+ return self._a2dp.interact(
+ test, interaction, description, pts_address)
+
+ # Handles unsupported profiles.
+ code = format_proxy(profile, interaction, description)
+ error_msg = (
+ f'Missing {profile} proxy and mmi: {interaction}\n'
+ f'Create a {profile.lower()}.py in mmi2grpc/:\n\n{code}\n'
+ f'Then, instantiate the corresponding proxy in __init__.py\n'
+ f'Finally, create a {profile.lower()}.proto in proto/pandora/'
+ f'and generate the corresponding interface.')
+
+ assert False, error_msg
diff --git a/mmi2grpc/_audio.py b/mmi2grpc/_audio.py
new file mode 100644
index 0000000..8e83c67
--- /dev/null
+++ b/mmi2grpc/_audio.py
@@ -0,0 +1,107 @@
+# 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.
+
+"""Audio tools."""
+
+import itertools
+import math
+import os
+from threading import Thread
+
+import numpy as np
+from scipy.io import wavfile
+
+SINE_FREQUENCY = 440
+SINE_DURATION = 0.1
+
+# File which stores the audio signal output data (after transport).
+# Used for running comparisons with the generated audio signal.
+OUTPUT_WAV_FILE = '/tmp/audiodata'
+
+WAV_RIFF_SIZE_OFFSET = 4
+WAV_DATA_SIZE_OFFSET = 40
+
+
+def _fixup_wav_header(path):
+ with open(path, 'r+b') as f:
+ f.seek(0, os.SEEK_END)
+ file_size = f.tell()
+ for offset in [WAV_RIFF_SIZE_OFFSET, WAV_DATA_SIZE_OFFSET]:
+ size = file_size - offset - 4
+ f.seek(offset)
+ f.write(size.to_bytes(4, byteorder='little'))
+
+
+class AudioSignal:
+ """Audio signal generator and verifier."""
+
+ def __init__(self, transport, amplitude, fs):
+ """Init AudioSignal class.
+
+ Args:
+ transport: function to send the generated audio data to.
+ amplitude: amplitude of the signal to generate.
+ fs: sampling rate of the signal to generate.
+ """
+ self.transport = transport
+ self.amplitude = amplitude
+ self.fs = fs
+ self.thread = None
+
+ def start(self):
+ """Generates the audio signal and send it to the transport."""
+ self.thread = Thread(target=self._run)
+ self.thread.start()
+
+ def _run(self):
+ sine = self._generate_sine(SINE_FREQUENCY, SINE_DURATION)
+
+ # Interleaved audio.
+ stereo = np.zeros(sine.size * 2, dtype=sine.dtype)
+ stereo[0::2] = sine
+
+ # Send 4 second of audio.
+ audio = itertools.repeat(stereo.tobytes(), int(4 / SINE_DURATION))
+
+ self.transport(audio)
+
+ def _generate_sine(self, f, duration):
+ sine = self.amplitude * \
+ np.sin(2 * np.pi * np.arange(self.fs * duration) * (f / self.fs))
+ s16le = (sine * 32767).astype('<i2')
+ return s16le
+
+ def verify(self):
+ """Verifies that the audio signal is correctly output."""
+ assert self.thread is not None
+ self.thread.join()
+ self.thread = None
+
+ _fixup_wav_header(OUTPUT_WAV_FILE)
+
+ samplerate, data = wavfile.read(OUTPUT_WAV_FILE)
+ # Take one second of audio after the first second.
+ audio = data[samplerate:samplerate*2, 0].astype(np.float) / 32767
+ assert len(audio) == samplerate
+
+ spectrum = np.abs(np.fft.fft(audio))
+ frequency = np.fft.fftfreq(samplerate, d=1/samplerate)
+ amplitudes = spectrum / (samplerate/2)
+ index = np.where(frequency == SINE_FREQUENCY)
+ amplitude = amplitudes[index][0]
+
+ match_amplitude = math.isclose(
+ amplitude, self.amplitude, rel_tol=1e-03)
+
+ return match_amplitude
diff --git a/mmi2grpc/_helpers.py b/mmi2grpc/_helpers.py
new file mode 100644
index 0000000..4e34d59
--- /dev/null
+++ b/mmi2grpc/_helpers.py
@@ -0,0 +1,94 @@
+# 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.
+
+"""Helper functions.
+
+Facilitates the implementation of a new profile proxy or a PTS MMI.
+"""
+
+import functools
+import textwrap
+import unittest
+
+DOCSTRING_WIDTH = 80 - 8 # 80 cols - 8 indentation spaces
+
+
+def assert_description(f):
+ """Decorator which verifies the description of a PTS MMI implementation.
+
+ Asserts that the docstring of a function implementing a PTS MMI is the same
+ as the corresponding official MMI description.
+
+ Args:
+ f: function implementing a PTS MMI.
+
+ Raises:
+ AssertionError: the docstring of the function does not match the MMI
+ description.
+ """
+ @functools.wraps(f)
+ def wrapper(*args, **kwargs):
+ description = textwrap.fill(
+ kwargs['description'], DOCSTRING_WIDTH, replace_whitespace=False)
+ docstring = textwrap.dedent(f.__doc__ or '')
+
+ if docstring.strip() != description.strip():
+ print(f'Expected description of {f.__name__}:')
+ print(description)
+
+ # Generate AssertionError.
+ test = unittest.TestCase()
+ test.maxDiff = None
+ test.assertMultiLineEqual(
+ docstring.strip(),
+ description.strip(),
+ f'description does not match with function docstring of'
+ f'{f.__name__}')
+
+ return f(*args, **kwargs)
+ return wrapper
+
+
+def format_function(mmi_name, mmi_description):
+ """Returns the base format of a function implementing a PTS MMI."""
+ wrapped_description = textwrap.fill(
+ mmi_description, DOCSTRING_WIDTH, replace_whitespace=False)
+ return (
+ f'@assert_description\n'
+ f'def {mmi_name}(self, **kwargs):\n'
+ f' """\n'
+ f'{textwrap.indent(wrapped_description, " ")}\n'
+ f' """\n'
+ f'\n'
+ f' return "OK"\n')
+
+
+def format_proxy(profile, mmi_name, mmi_description):
+ """Returns the base format of a profile proxy including a given MMI."""
+ wrapped_function = textwrap.indent(
+ format_function(mmi_name, mmi_description), ' ')
+ return (
+ f'from mmi2grpc._helpers import assert_description\n'
+ f'from mmi2grpc._proxy import ProfileProxy\n'
+ f'\n'
+ f'from pandora.{profile.lower()}_grpc import {profile}\n'
+ f'\n'
+ f'\n'
+ f'class {profile}Proxy(ProfileProxy):\n'
+ f'\n'
+ f' def __init__(self, channel):\n'
+ f' super().__init__()\n'
+ f' self.{profile.lower()} = {profile}(channel)\n'
+ f'\n'
+ f'{wrapped_function}')
diff --git a/mmi2grpc/_proxy.py b/mmi2grpc/_proxy.py
new file mode 100644
index 0000000..8eb4bd8
--- /dev/null
+++ b/mmi2grpc/_proxy.py
@@ -0,0 +1,42 @@
+# 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.
+
+"""Profile proxy base module."""
+
+from mmi2grpc._helpers import format_function
+
+
+class ProfileProxy:
+ """Profile proxy base class."""
+
+ def interact(
+ self, test: str, mmi_name: str, mmi_description: str,
+ pts_addr: bytes):
+ """Translate a MMI call to its corresponding implementation.
+
+ Args:
+ test: PTS test id.
+ mmi_name: MMI name.
+ mmi_description: MMI description.
+ pts_addr: Bluetooth MAC address of the PTS in bytes.
+
+ Raises:
+ AttributeError: the MMI is not implemented.
+ """
+ try:
+ return getattr(self, mmi_name)(
+ test=test, description=mmi_description, pts_addr=pts_addr)
+ except AttributeError:
+ code = format_function(mmi_name, mmi_description)
+ assert False, f'Unhandled mmi {id}\n{code}'
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"
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..de54399
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,15 @@
+[project]
+name = "mmi2grpc"
+authors = [{name = "Pandora", email = "pandora-core@google.com"}]
+readme = "README.md"
+dynamic = ["version", "description"]
+dependencies = [
+ "bt-test-interfaces",
+ "grpcio>=1.41",
+ "numpy>=1.22",
+ "scipy>=1.8"
+]
+
+[build-system]
+requires = ["flit_core==3.7.1"]
+build-backend = "flit_core.buildapi"
